CVE-2024-41129: Insufficient Information (NVD-CWE-noinfo)
value: The ops library is a Python framework for developing and testing Kubernetes and machine charms. The issue here is that ops passes the secret content as one of the args via CLI. This issue may affect any of the charms that are using: Juju (>=3.0), Juju secrets and not correctly capturing and processing `subprocess.CalledProcessError`. This vulnerability is fixed in 2.15.0.
CWE description: There is insufficient information about the issue to classify it; details are unkown or unspecified.
Published: 2024-07-22T15:15Z
Repository: https://github.com/canonical/operator
Language: Python
Files changed: ops/model.py, test/test_helpers.py, test/test_model.py
Commit date: 2024-07-22 09:45:52 +0900
Commit message: fix: use temp dir for secret data (#1290)

Change the mechanism used to pass secret values from ops to hook tools
from command line to files in temporary directory.

Removes the 2MB total secret size limit imposed by command line length

Hook tool calls juju agent which in turn reads these files, I've
confirmed with the Juju team that juju agent and charm code are run as
the same user.
Diff: diff --git a/ops/model.py b/ops/model.py
index de9b93e..3a9a1cf 100644
--- a/ops/model.py
+++ b/ops/model.py
@@ -3596,15 +3596,17 @@ class _ModelBackend:
             args.extend(['--description', description])
         if expire is not None:
             args.extend(['--expire', expire.isoformat()])
         if rotate is not None:
             args += ['--rotate', rotate.value]
-        if content is not None:
-            # The content has already been validated with Secret._validate_content
-            for k, v in content.items():
-                args.append(f'{k}={v}')
-        self._run_for_secret('secret-set', *args)
+        with tempfile.TemporaryDirectory() as tmp:
+            # The content is None or has already been validated with Secret._validate_content
+            for k, v in (content or {}).items():
+                with open(f'{tmp}/{k}', mode='w', encoding='utf-8') as f:
+                    f.write(v)
+                args.append(f'{k}#file={tmp}/{k}')
+            self._run_for_secret('secret-set', *args)
 
     def secret_add(
         self,
         content: Dict[str, str],
         *,
@@ -3623,14 +3625,17 @@ class _ModelBackend:
             args.extend(['--expire', expire.isoformat()])
         if rotate is not None:
             args += ['--rotate', rotate.value]
         if owner is not None:
             args += ['--owner', owner]
-        # The content has already been validated with Secret._validate_content
-        for k, v in content.items():
-            args.append(f'{k}={v}')
-        result = self._run('secret-add', *args, return_output=True)
+        with tempfile.TemporaryDirectory() as tmp:
+            # The content has already been validated with Secret._validate_content
+            for k, v in content.items():
+                with open(f'{tmp}/{k}', mode='w', encoding='utf-8') as f:
+                    f.write(v)
+                args.append(f'{k}#file={tmp}/{k}')
+            result = self._run('secret-add', *args, return_output=True)
         secret_id = typing.cast(str, result)
         return secret_id.strip()
 
     def secret_grant(self, id: str, relation_id: int, *, unit: Optional[str] = None):
         args = [id, '--relation', str(relation_id)]
diff --git a/test/test_helpers.py b/test/test_helpers.py
index 393fd75..84f9b20 100644
--- a/test/test_helpers.py
+++ b/test/test_helpers.py
@@ -151,10 +151,19 @@ class FakeScript:
             # Before executing the provided script, dump the provided arguments in calls.txt.
             # RS 'record separator' (octal 036 in ASCII), FS 'file separator' (octal 034 in ASCII).
             f.write(
                 """#!/bin/sh
 {{ printf {name}; printf "\\036%s" "$@"; printf "\\034"; }} >> {path}/calls.txt
+
+# Capture key and data from key#file=/some/path arguments
+for word in "$@"; do
+  echo "$word" | grep -q "#file=" || continue
+  key=$(echo "$word" | cut -d'#' -f1)
+  path=$(echo "$word" | cut -d'=' -f2)
+  cp "$path" "{path}/$key.secret"
+done
+
 {content}""".format_map(template_args)
             )
         path.chmod(0o755)
         # TODO: this hardcodes the path to bash.exe, which works for now but might
         #       need to be set via environ or something like that.
@@ -173,10 +182,13 @@ class FakeScript:
             calls = [line.split('\036') for line in f.read().split('\034')[:-1]]
             if clear:
                 f.truncate(0)
         return calls
 
+    def secrets(self) -> typing.Dict[str, str]:
+        return {p.stem: p.read_text() for p in self.path.iterdir() if p.suffix == '.secret'}
+
 
 class FakeScriptTest(unittest.TestCase):
     def test_fake_script_works(self):
         fake_script(self, 'foo', 'echo foo runs')
         fake_script(self, 'bar', 'echo bar runs')
diff --git a/test/test_model.py b/test/test_model.py
index b0e4f62..0100570 100644
--- a/test/test_model.py
+++ b/test/test_model.py
@@ -22,11 +22,11 @@ import re
 import tempfile
 import typing
 import unittest
 from collections import OrderedDict
 from textwrap import dedent
-from unittest.mock import MagicMock, patch
+from unittest.mock import ANY, MagicMock, patch
 
 import pytest
 
 import ops
 import ops.testing
@@ -3347,11 +3347,12 @@ class TestSecrets:
         secret = model.app.add_secret({'foo': 'x'})
         assert isinstance(secret, ops.Secret)
         assert secret.id == 'secret:123'
         assert secret.label is None
 
-        assert fake_script.calls(clear=True) == [['secret-add', '--owner', 'application', 'foo=x']]
+        assert fake_script.calls(clear=True) == [['secret-add', '--owner', 'application', ANY]]
+        assert fake_script.secrets() == {'foo': 'x'}
 
     def test_app_add_secret_args(self, fake_script: FakeScript, model: ops.Model):
         fake_script.write('secret-add', 'echo secret:234')
 
         expire = datetime.datetime(2022, 12, 9, 16, 17, 0)
@@ -3377,24 +3378,26 @@ class TestSecrets:
                 '2022-12-09T16:17:00',
                 '--rotate',
                 'hourly',
                 '--owner',
                 'application',
-                'foo=x',
-                'bar=y',
+                ANY,
+                ANY,
             ]
         ]
+        assert fake_script.secrets() == {'foo': 'x', 'bar': 'y'}
 
     def test_unit_add_secret_simple(self, fake_script: FakeScript, model: ops.Model):
         fake_script.write('secret-add', 'echo secret:345')
 
         secret = model.unit.add_secret({'foo': 'x'})
         assert isinstance(secret, ops.Secret)
         assert secret.id == 'secret:345'
         assert secret.label is None
 
-        assert fake_script.calls(clear=True) == [['secret-add', '--owner', 'unit', 'foo=x']]
+        assert fake_script.calls(clear=True) == [['secret-add', '--owner', 'unit', ANY]]
+        assert fake_script.secrets() == {'foo': 'x'}
 
     def test_unit_add_secret_args(self, fake_script: FakeScript, model: ops.Model):
         fake_script.write('secret-add', 'echo secret:456')
 
         expire = datetime.datetime(2022, 12, 9, 16, 22, 0)
@@ -3420,14 +3423,15 @@ class TestSecrets:
                 '2022-12-09T16:22:00',
                 '--rotate',
                 'yearly',
                 '--owner',
                 'unit',
-                'foo=w',
-                'bar=z',
+                ANY,
+                ANY,
             ]
         ]
+        assert fake_script.secrets() == {'foo': 'w', 'bar': 'z'}
 
     def test_unit_add_secret_errors(self, model: ops.Model):
         # Additional add_secret tests are done in TestApplication
         errors: typing.Any = [
             ({'xy': 'bar'}, {}, ValueError),
@@ -3719,14 +3723,15 @@ class TestSecretClass:
 
         with pytest.raises(ValueError):
             secret.set_content({'s': 't'})  # ensure it validates content (key too short)
 
         assert fake_script.calls(clear=True) == [
-            ['secret-set', 'secret:x', 'foo=bar'],
+            ['secret-set', 'secret:x', ANY],
             ['secret-info-get', '--label', 'y', '--format=json'],
-            ['secret-set', 'secret:z', 'bar=foo'],
+            ['secret-set', 'secret:z', ANY],
         ]
+        assert fake_script.secrets() == {'foo': 'bar', 'bar': 'foo'}
 
     def test_set_info(self, model: ops.Model, fake_script: FakeScript):
         fake_script.write('secret-set', """exit 0""")
         fake_script.write('secret-info-get', """echo '{"z": {"label": "y", "revision": 7}}'""")
 

Vulnerable code:         if content is not None:
            # The content has already been validated with Secret._validate_content
            for k, v in content.items():
                args.append(f'{k}={v}')
        self._run_for_secret('secret-set', *args)
        # The content has already been validated with Secret._validate_content
        for k, v in content.items():
            args.append(f'{k}={v}')
        result = self._run('secret-add', *args, return_output=True)
from unittest.mock import MagicMock, patch
        assert fake_script.calls(clear=True) == [['secret-add', '--owner', 'application', 'foo=x']]
                'foo=x',
                'bar=y',
        assert fake_script.calls(clear=True) == [['secret-add', '--owner', 'unit', 'foo=x']]
                'foo=w',
                'bar=z',
            ['secret-set', 'secret:x', 'foo=bar'],
            ['secret-set', 'secret:z', 'bar=foo'],
Fixed code:         with tempfile.TemporaryDirectory() as tmp:
            # The content is None or has already been validated with Secret._validate_content
            for k, v in (content or {}).items():
                with open(f'{tmp}/{k}', mode='w', encoding='utf-8') as f:
                    f.write(v)
                args.append(f'{k}#file={tmp}/{k}')
            self._run_for_secret('secret-set', *args)
        with tempfile.TemporaryDirectory() as tmp:
            # The content has already been validated with Secret._validate_content
            for k, v in content.items():
                with open(f'{tmp}/{k}', mode='w', encoding='utf-8') as f:
                    f.write(v)
                args.append(f'{k}#file={tmp}/{k}')
            result = self._run('secret-add', *args, return_output=True)

# Capture key and data from key#file=/some/path arguments
for word in "$@"; do
  echo "$word" | grep -q "#file=" || continue
  key=$(echo "$word" | cut -d'#' -f1)
  path=$(echo "$word" | cut -d'=' -f2)
  cp "$path" "{path}/$key.secret"
done

    def secrets(self) -> typing.Dict[str, str]:
        return {p.stem: p.read_text() for p in self.path.iterdir() if p.suffix == '.secret'}

from unittest.mock import ANY, MagicMock, patch
        assert fake_script.calls(clear=True) == [['secret-add', '--owner', 'application', ANY]]
        assert fake_script.secrets() == {'foo': 'x'}
                ANY,
                ANY,
        assert fake_script.secrets() == {'foo': 'x', 'bar': 'y'}
        assert fake_script.calls(clear=True) == [['secret-add', '--owner', 'unit', ANY]]
        assert fake_script.secrets() == {'foo': 'x'}
                ANY,
                ANY,
        assert fake_script.secrets() == {'foo': 'w', 'bar': 'z'}
            ['secret-set', 'secret:x', ANY],
            ['secret-set', 'secret:z', ANY],
        assert fake_script.secrets() == {'foo': 'bar', 'bar': 'foo'}