working to preserve SOPS-ness of stored data.

Author: carl-adams-planet

docs/changelog.md

@@ -1,5 +1,9 @@
 # Changelog
 
+## 2.3.1 - 2025-12-XX
+- Fix a bug where sops protected files would be rewritten without preserving
+  their sops protection.
+
 ## 2.3.0 - 2025-10-20
 - Improve the user experience around old stale sessions that appear to be
   initialized, but are actually expired.  This is done by providing the new

src/planet_auth/storage_utils.py

@@ -19,6 +19,7 @@
 import subprocess
 import time
 from abc import ABC, abstractmethod
+from enum import Enum
 from typing import Optional, Dict, Any
 
 from planet_auth.auth_exception import AuthException
@@ -96,9 +97,15 @@ def _default_storage_provider():
 class _SOPSAwareFilesystemObjectStorageProvider(ObjectStorageProvider):
     """
     Storage provider geared around backing a single object in a single file
-    with paths take from the root of the local file system.
+    with paths taken from the root of the local file system.
     """
 
+    _STORAGE_TYPE_KEY = "___SOPSAwareFilesystemObjectStorageProvider__storage_type"
+
+    class _StorageType(Enum):
+        PLAINTEXT = "plaintext"
+        SOPS = "sops"
+
     def __init__(self, root: Optional[pathlib.Path] = None):
         if root:
             self._storage_root = root
@@ -116,13 +123,34 @@ def _obj_filepath(self, obj_key):
         return obj_path
 
     @staticmethod
-    def _is_sops_path(file_path):
+    def _is_sops_path(file_path: pathlib.Path) -> bool:
         # TODO: Could be ".json.sops", or ".sops.json", depending on file
         #   level or field level encryption, respectively.  We currently
         #   only look for and support field level encryption in json
         #   files with a ".sops.json" suffix.
         return bool(file_path.suffixes == [".sops", ".json"])
 
+    # @staticmethod
+    # def _convert_to_sops_path(file_path: pathlib.Path) -> pathlib.Path:
+    #     if _SOPSAwareFilesystemObjectStorageProvider._is_sops_path(file_path):
+    #         return file_path
+    #     else:
+    #         if bool(file_path.suffix == ".json"):
+    #             return file_path.with_name(file_path.stem + ".sops" + file_path.suffix)
+    #         else:
+    #            return file_path.with_suffix(file_path.suffix + ".sops.json")
+
+    @staticmethod
+    def _filter_write_object(data: dict) -> dict:
+        # TODO: consider making this part of the base class?
+        final_data = {
+            k: v
+            for k, v in data.items()
+            # if k not in [_SOPSAwareFilesystemObjectStorageProvider._STORAGE_TYPE_KEY]
+            if not (isinstance(k, str) and k.startswith("__"))
+        }
+        return final_data
+
     @staticmethod
     def _read_json(file_path: pathlib.Path):
         auth_logger.debug(msg="Loading JSON data from file {}".format(file_path))
@@ -137,7 +165,7 @@ def _read_json_sops(file_path: pathlib.Path):
 
     @staticmethod
     def _write_json(file_path: pathlib.Path, data: dict):
-        auth_logger.debug(msg="Writing JSON data to file {}".format(file_path))
+        auth_logger.debug(msg="Writing JSON data to cleartext file {}".format(file_path))
         with open(file_path, mode="w", encoding="UTF-8") as file_w:
             os.chmod(file_path, stat.S_IREAD | stat.S_IWRITE)
             _no_none_data = {key: value for key, value in data.items() if value is not None}
@@ -155,26 +183,53 @@ def _write_json_sops(file_path: pathlib.Path, data: dict):
         #         ['sops', '-e', '--input-type', 'json', '--output-type',
         #          'json', '--output', file_path, '/dev/stdin'],
         #         stdin=data_f)
-        auth_logger.debug(msg="Writing JSON data to SOPS encrypted file {}".format(file_path))
         _SOPSAwareFilesystemObjectStorageProvider._write_json(file_path, data)
+        auth_logger.debug(msg="Writing JSON data to SOPS encrypted file {}".format(file_path))
         subprocess.check_call(["sops", "-e", "--input-type", "json", "--output-type", "json", "-i", file_path])
 
     @staticmethod
     def _load_file(file_path: pathlib.Path) -> dict:
         if _SOPSAwareFilesystemObjectStorageProvider._is_sops_path(file_path):
             new_data = _SOPSAwareFilesystemObjectStorageProvider._read_json_sops(file_path)
+            new_data[_SOPSAwareFilesystemObjectStorageProvider._STORAGE_TYPE_KEY] = (
+                _SOPSAwareFilesystemObjectStorageProvider._StorageType.SOPS.value
+            )
         else:
             new_data = _SOPSAwareFilesystemObjectStorageProvider._read_json(file_path)
+            new_data[_SOPSAwareFilesystemObjectStorageProvider._STORAGE_TYPE_KEY] = (
+                _SOPSAwareFilesystemObjectStorageProvider._StorageType.PLAINTEXT.value
+            )
 
         return new_data
 
+    @staticmethod
+    def _do_sops(file_path: pathlib.Path, data: dict) -> bool:
+        if _SOPSAwareFilesystemObjectStorageProvider._is_sops_path(file_path):
+            return True
+        if (
+            data
+            and data.get(_SOPSAwareFilesystemObjectStorageProvider._STORAGE_TYPE_KEY)
+            == _SOPSAwareFilesystemObjectStorageProvider._StorageType.SOPS.value
+        ):
+            auth_logger.warning(msg=f"Data sourced from SOPS being written cleartext to the file {file_path}.")
+            # Upgrading to SOPS would be great, but also problematic.
+            # return True
+
+        return False
+
     @staticmethod
     def _save_file(file_path: pathlib.Path, data: dict):
         file_path.parent.mkdir(parents=True, exist_ok=True)
-        if _SOPSAwareFilesystemObjectStorageProvider._is_sops_path(file_path):
-            _SOPSAwareFilesystemObjectStorageProvider._write_json_sops(file_path, data)
+        do_sops = _SOPSAwareFilesystemObjectStorageProvider._do_sops(file_path, data)
+        write_data = _SOPSAwareFilesystemObjectStorageProvider._filter_write_object(data)
+
+        if do_sops:
+            # This has to be with the caller, otherwise the caller would not know
+            # where we actually wrote the data and would likely not be able to find it again.
+            # sops_file_path = _SOPSAwareFilesystemObjectStorageProvider._convert_to_sops_path(file_path)
+            _SOPSAwareFilesystemObjectStorageProvider._write_json_sops(file_path, write_data)
         else:
-            _SOPSAwareFilesystemObjectStorageProvider._write_json(file_path, data)
+            _SOPSAwareFilesystemObjectStorageProvider._write_json(file_path, write_data)
 
     def load_obj(self, key: ObjectStorageProvider_KeyType) -> dict:
         obj_filepath = self._obj_filepath(key)

src/planet_auth_utils/commands/cli/main.py

@@ -227,8 +227,7 @@ def cmd_plauth_login(
     )
     print("Login succeeded.")  # Errors should throw.
 
-    # FIXME: sops
-    post_login_cmd_helper(override_auth_context=override_auth_context, use_sops=sops, prompt_pre_selection=yes)
+    post_login_cmd_helper(override_auth_context=override_auth_context, use_sops_opt=sops, prompt_pre_selection=yes)
 
 
 cmd_plauth.add_command(cmd_oauth)

src/planet_auth_utils/commands/cli/oauth_cmd.py

@@ -128,8 +128,7 @@ def cmd_oauth_login(
         extra=login_extra,
     )
     print("Login succeeded.")  # Errors should throw.
-    # FIXME sops-ness of profiles is being lost
-    post_login_cmd_helper(override_auth_context=current_auth_context, use_sops=sops, prompt_pre_selection=yes)
+    post_login_cmd_helper(override_auth_context=current_auth_context, use_sops_opt=sops, prompt_pre_selection=yes)
 
 
 @cmd_oauth.command("refresh")

src/planet_auth_utils/commands/cli/planet_legacy_auth_cmd.py

@@ -81,8 +81,7 @@ def cmd_pllegacy_login(ctx, username, password, sops, yes):
         password=password,
     )
     print("Login succeeded.")  # Errors should throw.
-    # FIXME: sops-ness is being lost
-    post_login_cmd_helper(override_auth_context=current_auth_context, use_sops=sops, prompt_pre_selection=yes)
+    post_login_cmd_helper(override_auth_context=current_auth_context, use_sops_opt=sops, prompt_pre_selection=yes)
 
 
 @cmd_pllegacy.command("print-api-key")

src/planet_auth_utils/commands/cli/util.py

@@ -17,14 +17,18 @@
 import json
 from typing import List, Optional
 
+import planet_auth.logging.auth_logger
 import planet_auth
 from planet_auth.constants import AUTH_CONFIG_FILE_SOPS, AUTH_CONFIG_FILE_PLAIN
+from planet_auth.storage_utils import _SOPSAwareFilesystemObjectStorageProvider
 from planet_auth.util import custom_json_class_dumper
 
 from planet_auth_utils.builtins import Builtins
 from planet_auth_utils.profile import Profile
 from .prompts import prompt_and_change_user_default_profile_if_different
 
+auth_logger = planet_auth.logging.auth_logger.getAuthLogger()
+
 
 def monkeypatch_hide_click_cmd_options(cmd, hide_options: List[str]):
     """
@@ -68,7 +72,7 @@ def print_obj(obj):
 
 
 def post_login_cmd_helper(
-    override_auth_context: planet_auth.Auth, use_sops, prompt_pre_selection: Optional[bool] = None
+    override_auth_context: planet_auth.Auth, use_sops_opt, prompt_pre_selection: Optional[bool] = None
 ):
     override_profile_name = override_auth_context.profile_name()
     if not override_profile_name:
@@ -88,6 +92,18 @@ def post_login_cmd_helper(
     # helping CLI commands, we can be more opinionated about what is to
     # be done.
 
+    # If the profile was read from sops, selected sops even if wasn't an
+    # explict command line option.
+    use_sops = use_sops_opt
+    if not use_sops:
+        _config_data = override_auth_context.auth_client().config().data() or {}
+        if (
+            _config_data.get(_SOPSAwareFilesystemObjectStorageProvider._STORAGE_TYPE_KEY)
+            == _SOPSAwareFilesystemObjectStorageProvider._StorageType.SOPS.value
+        ):
+            auth_logger.debug(msg="Implicitly selecting SOPS based on data originating from SOPS protected storage.")
+            use_sops = True
+
     # Don't clobber built-in profiles.
     if not Builtins.is_builtin_profile(override_profile_name):
         if use_sops:

version.txt

@@ -1 +1 @@
-2.3.0
+2.3.1