Implement offline access in python

openc3/lib/openc3/io/json_api_object.rb

@@ -57,7 +57,7 @@ class JsonApiObject
     # @param url [String] The url of openc3-cosmos-cmd-tlm-api http://openc3-cosmos-cmd-tlm-api:2901
     # @param timeout [Float] The time to wait before disconnecting 1.0
     # @param authentication [OpenC3Authentication] The authentication object if nill initialize will generate
-    def initialize(url: ENV['OPENC3_API_URL'], timeout: 1.0, authentication: nil)
+    def initialize(url:, timeout: 1.0, authentication: nil)
       @http = nil
       @mutex = Mutex.new
       @request_data = ""

openc3/lib/openc3/io/json_drb_object.rb

@@ -43,7 +43,7 @@ class JsonDRbObject < JsonApiObject
     # @param url [String] The url of openc3-cosmos-cmd-tlm-api http://openc3-cosmos-cmd-tlm-api:2901
     # @param timeout [Float] The time to wait before disconnecting 1.0
     # @param authentication [OpenC3Authentication] The authentication object if nill initialize will generate
-    def initialize(url: ENV['OPENC3_API_URL'], timeout: 1.0, authentication: nil)
+    def initialize(url:, timeout: 1.0, authentication: nil)
       super(url: url, timeout: timeout, authentication: authentication)
       @uri = URI("#{url}/openc3-api/api")
     end

openc3/lib/openc3/models/offline_access_model.rb

@@ -40,7 +40,7 @@ def self.all(scope:)
     # END NOTE
 
     def initialize(name:, offline_access_token: nil, updated_at: nil, scope:)
-      super("#{scope}__#{PRIMARY_KEY}", name: name, scope: scope)
+      super("#{scope}__#{PRIMARY_KEY}", name: name, updated_at: updated_at, scope: scope)
       @offline_access_token = offline_access_token
     end
 

openc3/python/openc3/api/offline_access_api.py

@@ -0,0 +1,73 @@
+# Copyright 2025 OpenC3, Inc
+# All Rights Reserved.
+#
+# This program is free software; you can modify and/or redistribute it
+# under the terms of the GNU Affero General Public License
+# as published by the Free Software Foundation; version 3 with
+# attribution addendums as found in the LICENSE.txt
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Affero General Public License for more details.
+
+# This file may also be used under the terms of a commercial license
+# if purchased from OpenC3, Inc.
+
+import os
+from openc3.api import WHITELIST
+from openc3.environment import OPENC3_SCOPE
+from openc3.utilities.authorization import authorize
+from openc3.models.offline_access_model import OfflineAccessModel
+from openc3.utilities.authentication import (
+    OpenC3KeycloakAuthentication,
+)
+try:
+    from openc3enterprise.utilities.authorization import user_info
+except ImportError:
+    from openc3.utilities.authorization import user_info
+
+WHITELIST.extend(["offline_access_needed", "set_offline_access"])
+
+
+def offline_access_needed(manual=False, scope=OPENC3_SCOPE, token=None):
+    authorize(permission="system", manual=manual, scope=scope)
+    try:
+        authorize(permission="script_view", manual=manual, scope=scope)
+    except:
+        # Not needed if can't run scripts
+        return False
+    info = user_info(token)
+    if "offline_access" in info["roles"]:
+        username = info["username"]
+        if username and username != "":
+            model = OfflineAccessModel.get_model(name=username, scope=scope)
+            if model and model.offline_access_token:
+                auth = OpenC3KeycloakAuthentication(os.environ.get("OPENC3_KEYCLOAK_URL"))
+                valid_token = auth.get_token_from_refresh_token(model.offline_access_token)
+                if valid_token:
+                    return False
+                else:
+                    model.offline_access_token = None
+                    model.update
+                    return True
+            return True
+        else:
+            return False
+    else:
+        return False
+
+
+def set_offline_access(offline_access_token, manual=False, scope=OPENC3_SCOPE, token=None):
+    authorize(permission="script_view", manual=manual, scope=scope)
+    info = user_info(token)
+    username = info["username"]
+    if not username or username == "":
+        raise "Invalid username"
+    model = OfflineAccessModel.get_model(name=username, scope=scope)
+    if model:
+        model.offline_access_token = offline_access_token
+        model.update
+    else:
+        model = OfflineAccessModel(name=username, offline_access_token=offline_access_token, scope=scope)
+        model.create

openc3/python/openc3/models/offline_access_model.py

@@ -0,0 +1,51 @@
+# Copyright 2025 OpenC3, Inc.
+# All Rights Reserved.
+#
+# This program is free software; you can modify and/or redistribute it
+# under the terms of the GNU Affero General Public License
+# as published by the Free Software Foundation; version 3 with
+# attribution addendums as found in the LICENSE.txt
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Affero General Public License for more details.
+
+# This file may also be used under the terms of a commercial license
+# if purchased from OpenC3, Inc.
+
+from openc3.environment import OPENC3_SCOPE
+from openc3.models.model import Model
+
+
+class OfflineAccessModel(Model):
+    PRIMARY_KEY = "openc3__offline_access"
+
+    # NOTE: The following three class methods are used by the ModelController
+    # and are reimplemented to enable various Model class methods to work
+    @classmethod
+    def get(cls, name: str, scope: str):
+        return super().get(f"{scope}{OfflineAccessModel.PRIMARY_KEY}", name=name)
+
+    @classmethod
+    def names(cls, scope: str):
+        return super().names(f"{scope}{OfflineAccessModel.PRIMARY_KEY}")
+
+    @classmethod
+    def all(cls, scope: str):
+        return super().all(f"{scope}{OfflineAccessModel.PRIMARY_KEY}")
+
+    def __init__(
+        self, name: str, offline_access_token: str = None, updated_at: float = None, scope: str = OPENC3_SCOPE
+    ):
+        super(f"{scope}__{OfflineAccessModel.PRIMARY_KEY}", name=name, updated_at=updated_at, scope=scope)
+        self.offline_access_token = offline_access_token
+
+    # @return [Hash] JSON encoding of this model
+    def as_json(self):
+        return {
+            "name": self.name,
+            "updated_at": self.updated_at,
+            "offline_access_token": self.offline_access_token,
+            "scope": self.scope,
+        }