tegwick
e7282e493d
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a0e6f1-443f-7783-9920-a16b2ffc467f
42 lines
2.4 KiB
Python
42 lines
2.4 KiB
Python
from dataclasses import replace
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
import unittest
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(ROOT / 'scripts'))
|
|
from alert_ack import Actor
|
|
from alert_policy import Policy
|
|
|
|
|
|
@unittest.skipUnless(os.environ.get('RTEL_FLEX_AUTH_BINARY'), 'set RTEL_FLEX_AUTH_BINARY')
|
|
class NativePolicyTests(unittest.TestCase):
|
|
def test_native_wire_digest_and_revoked_or_wrong_identity_denials(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
request_file = Path(directory) / 'request.json'
|
|
class Native:
|
|
def request(self, method, url, body=None, headers=None):
|
|
request_file.write_bytes(body)
|
|
result = subprocess.run([os.environ['RTEL_FLEX_AUTH_BINARY'], 'check', '--policy',
|
|
str(ROOT / 'integration/telemetry-policy.md'), '--registry', str(ROOT / 'integration/registry.json'),
|
|
'--request', str(request_file)], capture_output=True, check=True)
|
|
return 200, json.loads(result.stdout)
|
|
actor = Actor('https://kc.coulomb.social', 'uid=tegwick,ou=people,dc=netkingdom,dc=local',
|
|
'tenant:platform', ('railiance-admin',), time.time() + 600, 'c' * 32,
|
|
assurance=dict(level='aal2', mfa=True, source='key-cape', methods=['pwd', 'otp'], at=int(time.time())),
|
|
groups=('railiance-admins',), tenant_source='directory')
|
|
# Native digest is pinned by the owner packet, not inferred from a response.
|
|
policy = Policy('https://policy.example', lambda: 'fixture-only',
|
|
'railiance-telemetry.alert-acknowledgment', 'v1',
|
|
'sha256:02938202cf75140d6ab638b8d0fbce6ebb22832354efc89cab040b13e8943f09', transport=Native())
|
|
resource = 'alert:11111111-1111-1111-1111-111111111111'
|
|
self.assertEqual(policy(actor, 'acknowledge', resource)['effect'], 'allow')
|
|
for changed in (replace(actor, roles=()), replace(actor, groups=()), replace(actor, subject='other'),
|
|
replace(actor, tenant='tenant:other'), replace(actor, issuer='https://other.example'),
|
|
replace(actor, assurance=dict(actor.assurance, at=1))):
|
|
self.assertEqual(policy(changed, 'acknowledge', resource)['effect'], 'deny')
|