849f3947dc
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a17f16-d99c-7ea3-95f0-8212897fba6b
67 lines
3.1 KiB
Python
67 lines
3.1 KiB
Python
"""Synthetic access refusals; no live API or private custody values."""
|
|
import base64
|
|
import copy
|
|
import importlib.util
|
|
import sys
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
SCRIPTS = Path(__file__).resolve().parents[1] / "scripts"
|
|
sys.path.insert(0, str(SCRIPTS))
|
|
spec = importlib.util.spec_from_file_location("access", SCRIPTS / "application_recovery_access.py")
|
|
m = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(m)
|
|
|
|
|
|
class AccessTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.contract = {"schema": "platform.coulomb-recovery-api.v1", "project_id": m.PROJECT,
|
|
"materials": {"reader-pair": {"id": "synthetic-id", "name": "synthetic-name", "revision": 1}}}
|
|
self.metadata = {"id": "synthetic-id", "name": "synthetic-name", "project_id": m.PROJECT,
|
|
"protected": True, "type": "opaque", "ephemeral_policy": None}
|
|
self.version = {"status": "enabled", "ephemeral_properties": {}}
|
|
self.accesses = 0
|
|
|
|
def api(self, method, path):
|
|
self.assertEqual(method, "GET")
|
|
if path.endswith("/access"):
|
|
self.accesses += 1
|
|
return {"secret_id": "synthetic-id", "revision": 1, "data": base64.b64encode(
|
|
b'{"AWS_ACCESS_KEY_ID":"SYNTHETIC_ID","AWS_SECRET_ACCESS_KEY":"SYNTHETIC_VALUE"}').decode()}
|
|
if "/versions/" in path:
|
|
return self.version
|
|
return self.metadata
|
|
|
|
def test_routine_retrieval_uses_only_protected_bound_scoped_secret(self):
|
|
payload = m.retrieve(self.api, self.contract, "reader-pair")
|
|
self.assertIn(b"SYNTHETIC_ID", payload)
|
|
self.assertEqual(self.accesses, 1)
|
|
|
|
def test_project_name_protection_type_and_expiry_drift_refuse_before_data(self):
|
|
baseline = copy.deepcopy(self.metadata)
|
|
for key, value in (("project_id", "foreign"), ("name", "foreign"), ("protected", False),
|
|
("type", "certificate"), ("ephemeral_policy", {"time_to_live": "60s"}),
|
|
("deletion_requested_at", "synthetic-time")):
|
|
self.metadata = dict(baseline, **{key: value})
|
|
with self.assertRaises(m.Refused):
|
|
m.retrieve(self.api, self.contract, "reader-pair")
|
|
self.assertEqual(self.accesses, 0)
|
|
|
|
def test_disabled_or_expiring_version_refuses_before_data(self):
|
|
for version in ({"status": "disabled"}, {"status": "enabled", "ephemeral_properties": {"expires_at": "soon"}},
|
|
{"status": "enabled", "ephemeral_properties": {"expires_once_accessed": True}}):
|
|
self.version = version
|
|
with self.assertRaises(m.Refused):
|
|
m.retrieve(self.api, self.contract, "reader-pair")
|
|
self.assertEqual(self.accesses, 0)
|
|
|
|
def test_wrong_response_revision_refused(self):
|
|
def api(method, path):
|
|
result = self.api(method, path)
|
|
return dict(result, revision=2) if path.endswith("/access") else result
|
|
with self.assertRaises(m.Refused):
|
|
m.retrieve(api, self.contract, "reader-pair")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|