[orion-ci] Add 'append' and 'mask' options to file secrets.

services/orion-decision/pyproject.toml

@@ -51,8 +51,11 @@ filterwarnings = [
 ]
 
 [tool.ruff]
-ignore = []
 line-length = 88
+target-version = "py38"
+
+[tool.ruff.lint]
+ignore = []
 select = [
     # flake8
     "E",
@@ -65,7 +68,6 @@ select = [
     # includes yesqa
     "RUF",
 ]
-target-version = "py38"
 
-[tool.ruff.isort]
+[tool.ruff.lint.isort]
 known-first-party = ["orion_decision"]

services/orion-decision/src/orion_decision/ci_matrix.py

@@ -445,7 +445,7 @@ def __str__(self) -> str:
         return json_dumps(self.serialize())
 
     @abstractmethod
-    def serialize(self) -> Dict[str, Optional[str]]:
+    def serialize(self) -> Dict[str, Optional[Union[bool, str]]]:
         """Return a JSON serializable copy of self."""
 
     @staticmethod
@@ -467,7 +467,13 @@ def from_json(data: str) -> "CISecret":
         if obj["type"] == "env":
             return CISecretEnv(obj["secret"], obj["name"], key=obj.get("key"))
         if obj["type"] == "file":
-            return CISecretFile(obj["secret"], obj["path"], key=obj.get("key"))
+            return CISecretFile(
+                obj["secret"],
+                obj["path"],
+                key=obj.get("key"),
+                append=obj.get("append", False),
+                mask=obj.get("mask"),
+            )
         return CISecretKey(
             obj["secret"], hostname=obj.get("hostname"), key=obj.get("key")
         )
@@ -495,7 +501,7 @@ def __init__(self, secret: str, name: str, key: Optional[str] = None) -> None:
         super().__init__(secret, key)
         self.name = name
 
-    def serialize(self) -> Dict[str, Optional[str]]:
+    def serialize(self) -> Dict[str, Optional[Union[bool, str]]]:
         """Return a JSON serializable copy of self.
 
         Returns:
@@ -518,20 +524,31 @@ class CISecretFile(CISecret):
         (see CISecret for attributes defined there)
     """
 
-    __slots__ = ("path",)
+    __slots__ = ("path", "append", "mask")
 
-    def __init__(self, secret: str, path: str, key: Optional[str] = None) -> None:
+    def __init__(
+        self,
+        secret: str,
+        path: str,
+        key: Optional[str] = None,
+        append: bool = False,
+        mask: Optional[str] = None,
+    ) -> None:
         """Initialize CISecretFile object.
 
         Arguments:
             secret: Taskcluster namespace where the secret is held.
             path: Path where secret should be written to.
             key: Sub-key in the Taskcluster secret that contains the value.
+            append: Whether the file should be appended to, if it already exists.
+            mask: Permission mask to apply after writing file.
         """
         super().__init__(secret, key)
         self.path = path
+        self.append = append or False
+        self.mask = int(mask, 8) if mask is not None else None
 
-    def serialize(self) -> Dict[str, Optional[str]]:
+    def serialize(self) -> Dict[str, Optional[Union[bool, str]]]:
         """Return a JSON serializable copy of self.
 
         Returns:
@@ -542,6 +559,9 @@ def serialize(self) -> Dict[str, Optional[str]]:
             "key": self.key,
             "secret": self.secret,
             "path": self.path,
+            "append": self.append,
+            # serialize back to octal string as expected by schema
+            "mask": f"{self.mask:o}" if self.mask is not None else None,
         }
 
     def write(self) -> None:
@@ -552,8 +572,12 @@ def write(self) -> None:
         data = self.get_secret_data()
         if not isinstance(data, str):
             data = json_dumps(data)
-        Path(self.path).parent.mkdir(parents=True, exist_ok=True)
-        Path(self.path).write_text(data)
+        dest = Path(self.path)
+        dest.parent.mkdir(parents=True, exist_ok=True)
+        with dest.open("a" if self.append else "w") as secret_fp:
+            secret_fp.write(data)
+        if self.mask is not None:
+            dest.chmod(self.mask)
 
 
 class CISecretKey(CISecret):
@@ -580,7 +604,7 @@ def __init__(
         super().__init__(secret, key)
         self.hostname = hostname
 
-    def serialize(self) -> Dict[str, Optional[str]]:
+    def serialize(self) -> Dict[str, Optional[Union[bool, str]]]:
         """Return a JSON serializable copy of self.
 
         Returns:

services/orion-decision/src/orion_decision/schemas/ci_secret.yaml

@@ -35,6 +35,14 @@ oneOf:
       path:
         type: string
         minLength: 1
+      append:
+        type: boolean
+        description: Append to existing file (or create if not exist)
+      mask:
+        type: string
+        minLength: 1
+        pattern: "^[0-7]{1,4}?$"
+        description: Set the file permission bits using `chmod`
     properties:
       type:
         const: file

services/orion-decision/tox.ini

@@ -3,7 +3,7 @@ envlist = py{38,39,310,311,312},lint
 skip_missing_interpreters = true
 tox_pip_extensions_ext_venv_update = true
 
-[testenv:py{38,39,310,311}]
+[testenv:py{38,39,310,311,312}]
 usedevelop = true
 deps =
     freezegun
@@ -18,10 +18,10 @@ commands = orion-check {posargs}
 
 [testenv:lint]
 deps =
-    black==v23.11.0
-    mypy==v1.7
+    black==v24.4.2
+    mypy==v1.11.0
     pytest-mock
-    ruff==v0.1.5
+    ruff==v0.5.5
 usedevelop = true
 allowlist_externals =
     bash