railiance-platform/tests/test_informed_decision_requester_metadata.py

import contextlib
import importlib.util
import io
import json
from pathlib import Path
import subprocess
import unittest
from unittest.mock import patch

spec = importlib.util.spec_from_file_location('requester_metadata', Path(__file__).resolve().parents[1] / 'scripts/informed_decision_requester_metadata.py')
m = importlib.util.module_from_spec(spec)
spec.loader.exec_module(m)


def role(**updates):
    return dict(bound_service_account_names=['informed-decision-eso'],
                bound_service_account_namespaces=['informed-decision'], audience='openbao',
                token_policies=[m.POLICY], token_ttl=900, token_max_ttl=900,
                token_explicit_max_ttl=900, token_no_default_policy=True, **updates)


class MetadataTests(unittest.TestCase):
    def test_role_rejects_widening_and_unbounded_token(self):
        original = role()
        for key, value in [('bound_service_account_names', ['*']),
                           ('bound_service_account_namespaces', ['*']),
                           ('audience', 'other'), ('token_policies', [m.POLICY, 'default']),
                           ('token_ttl', 3600), ('token_max_ttl', 0),
                           ('token_explicit_max_ttl', 0), ('token_no_default_policy', 1),
                           ('token_period', 900), ('bound_service_account_namespace_selector', '*')]:
            with self.subTest(key=key):
                self.assertFalse(m.role_matches(dict(original, **{key: value})))
        self.assertTrue(m.role_matches(original))

    def test_differing_role_refuses_before_missing_policy_write(self):
        calls = []
        def command(args, payload=None, **kwargs):
            calls.append(args)
            return None if args[-1].startswith('sys/') else {'data': dict(role(), audience='wrong')}
        with self.assertRaises(m.Refused) as caught:
            m.apply('exact policy', {}, command=command)
        self.assertEqual(caught.exception.code, 3)
        self.assertTrue(all(args[0] == 'read' for args in calls))

    def test_differing_policy_refuses_all_writes(self):
        calls = []
        def command(args, payload=None, **kwargs):
            calls.append(args)
            return {'data': {'policy': 'different'}} if args[-1].startswith('sys/') else None
        with self.assertRaises(m.Refused) as caught:
            m.apply('exact policy', {}, command=command)
        self.assertEqual(caught.exception.code, 3)
        self.assertTrue(all(args[0] == 'read' for args in calls))

    def test_identical_objects_noop_without_write(self):
        calls = []
        def command(args, payload=None, **kwargs):
            calls.append(args)
            return {'data': {'policy': 'exact policy'}} if args[-1].startswith('sys/') else {'data': role()}
        m.apply('exact policy', {}, command=command)
        self.assertEqual(len(calls), 4)
        self.assertTrue(all(args[0] == 'read' for args in calls))

    def test_new_objects_write_only_two_metadata_paths_then_verify(self):
        calls = []
        saved = {}
        def command(args, payload=None, **kwargs):
            calls.append(args)
            path = args[2]
            if args[0] == 'write':
                saved[path] = json.loads(payload)
                return {}
            if path not in saved:
                return None
            return {'data': saved[path]} if path.startswith('sys/') else {'data': role()}
        m.apply('exact policy', {'audience': 'openbao'}, command=command)
        self.assertEqual(set(saved), {'sys/policies/acl/' + m.POLICY, m.ROLE})
        self.assertEqual([args[0] for args in calls], ['read', 'read', 'write', 'write', 'read', 'read'])

    def test_postwrite_role_drift_fails_verification(self):
        calls = []
        def command(args, payload=None, **kwargs):
            calls.append(args)
            if kwargs.get('missing_ok'): return None
            if args[0] == 'write': return {}
            return {'data': {'policy': 'exact policy'}} if args[-1].startswith('sys/') else {'data': dict(role(), audience='wrong')}
        with self.assertRaises(m.Refused) as caught:
            m.apply('exact policy', {}, command=command)
        self.assertEqual(caught.exception.code, 5)

    def test_404_only_is_absence_and_failures_are_silent(self):
        for code, stderr, missing in [(2, b'Code: 404', True), (2, b'Code: 403 SECRET-SENTINEL', False), (1, b'Code: 404 SECRET-SENTINEL', False)]:
            with self.subTest(code=code, stderr=stderr), patch.object(m.subprocess, 'run', return_value=subprocess.CompletedProcess([], code, b'SECRET-SENTINEL', stderr)):
                output = io.StringIO()
                with contextlib.redirect_stdout(output), contextlib.redirect_stderr(output):
                    if missing: self.assertIsNone(m.cli(['read', '-format=json', m.ROLE], missing_ok=True))
                    else:
                        with self.assertRaises(m.Refused): m.cli(['read', '-format=json', m.ROLE], missing_ok=True)
                self.assertEqual(output.getvalue(), '')

    def test_empty_204_write_is_success(self):
        with patch.object(m.subprocess, 'run', return_value=subprocess.CompletedProcess([], 0, b'', b'')):
            self.assertEqual(m.cli(['write', '-format=json', m.ROLE, '-'], b'{}'), {})

    def test_proposed_source_refuses_before_network(self):
        digest = m.hashlib.sha256(m.CCR.read_bytes()).hexdigest()
        with patch.object(m.subprocess, 'run', side_effect=AssertionError('must remain offline')):
            with self.assertRaises(m.Refused) as caught: m.source(digest)
        self.assertEqual(caught.exception.code, 2)

    def test_missing_arguments_silent(self):
        with patch.object(m.sys, 'argv', ['metadata-child']):
            output = io.StringIO()
            with contextlib.redirect_stdout(output), contextlib.redirect_stderr(output):
                self.assertEqual(m.main(), 2)
            self.assertEqual(output.getvalue(), '')


if __name__ == '__main__':
    unittest.main()