security: reject Playwright audience and delivery alias collisions
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a1346b-72a4-7d40-b8bd-8b8ca346cfa2
This commit is contained in:
parent
eba6e1fdbb
commit
276b95f6ba
6 changed files with 33 additions and 2 deletions
|
|
@ -107,6 +107,11 @@ reviewed source revision. Its result contains the private original registry and
|
|||
must remain inside that execution path. Only `summary` is a public receipt.
|
||||
Source revision and digest bindings do not constitute an approval or custody grant.
|
||||
|
||||
Inventory checks use the issuer's effective audience, including its client-ID
|
||||
fallback when `audience` is absent/empty. A second issuer environment variable
|
||||
pointing to the same proposed agent Secret/key is refused even if its name differs.
|
||||
These checks do not replace Platform's wider custody/consumer isolation review.
|
||||
|
||||
The adapter admits one replica, `Recreate`, the pinned immutable KeyCape image,
|
||||
the expected registry mount and explicit environment bindings. It preserves
|
||||
unrelated settings and refuses differing bindings. `check_registry` pins the
|
||||
|
|
|
|||
|
|
@ -83,6 +83,12 @@ def build(secret, deployment, identities, source_revision):
|
|||
records = registration.clients(identities)
|
||||
for record in records[1:]:
|
||||
binding = registration.registry.env_binding(record)
|
||||
target_ref = binding['valueFrom']['secretKeyRef']
|
||||
for entry in proposed_env:
|
||||
reference = entry.get('valueFrom', {}).get('secretKeyRef', {})
|
||||
require(entry['name'] == binding['name'] or
|
||||
(reference.get('name'), reference.get('key')) != (target_ref['name'], target_ref['key']),
|
||||
'delivery_secret_already_bound')
|
||||
existing = [e for e in proposed_env if e['name'] == binding['name']]
|
||||
require(not existing or existing == [binding], 'existing_binding_requires_reconciliation')
|
||||
if not existing:
|
||||
|
|
|
|||
|
|
@ -100,14 +100,14 @@ def prepare_registry(secret, identities, operation='add'):
|
|||
require(isinstance(config.get('clients'), list), 'client_sequence_required')
|
||||
for existing in config['clients']:
|
||||
require(isinstance(existing, dict), 'client_shape_unexpected')
|
||||
if existing.get('audience') == AUDIENCE:
|
||||
if (existing.get('audience') or existing.get('clientId')) == AUDIENCE:
|
||||
require(existing in wanted, 'unreviewed_playwright_audience_registration')
|
||||
# Preflight every record against the same original, including collisions.
|
||||
for client in wanted:
|
||||
_, current, _ = registry._inspect(secret, client)
|
||||
if 'serviceSubject' in client:
|
||||
for other in current['clients']:
|
||||
if other['clientId'] != client['clientId'] and other.get('audience') == AUDIENCE:
|
||||
if other['clientId'] != client['clientId'] and (other.get('audience') or other.get('clientId')) == AUDIENCE:
|
||||
require(other.get('serviceSubject') != client['serviceSubject'],
|
||||
'playwright_subject_already_registered')
|
||||
matching = [c for c in current['clients'] if c['clientId'] == client['clientId']]
|
||||
|
|
|
|||
|
|
@ -217,5 +217,12 @@ class ActivationTests(unittest.TestCase):
|
|||
with self.assertRaisesRegex(a.Refused, '^native_verifier_digest_mismatch$'):
|
||||
a.check_registry(self.proposal, binary, 'wrong')
|
||||
|
||||
def test_other_env_name_cannot_alias_a_proposed_agent_secret(self):
|
||||
binding = a.registration.registry.env_binding(a.registration.agent_client(IDENTITIES[0]))
|
||||
binding['name'] = 'OTHER_CLIENT_SECRET'
|
||||
a.container(self.deployment['spec'])['env'].append(binding)
|
||||
with self.assertRaisesRegex(a.Refused, 'delivery_secret_already_bound'):
|
||||
a.build(self.secret, self.deployment, IDENTITIES, REVISION)
|
||||
|
||||
|
||||
if __name__ == '__main__': unittest.main()
|
||||
|
|
|
|||
|
|
@ -119,6 +119,14 @@ class RegistrationTests(unittest.TestCase):
|
|||
encoded, _ = p.prepare_registry(secret, [IDENTITY])
|
||||
self.assertIn(existing, yaml.safe_load(decode(encoded))['clients'])
|
||||
|
||||
def test_default_audience_client_id_cannot_bypass_dedicated_inventory(self):
|
||||
base, _ = fixture()
|
||||
default_audience = dict(p.public_client(), clientId=p.AUDIENCE)
|
||||
del default_audience['audience']
|
||||
secret = altered(base, lambda c: c['clients'].append(default_audience))
|
||||
with self.assertRaises(p.Refused):
|
||||
p.prepare_registry(secret, [IDENTITY])
|
||||
|
||||
def test_cli_outputs_only_proposal_and_no_false_admission(self):
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
path = Path(directory) / 'agents.json'; path.write_text(json.dumps([IDENTITY]))
|
||||
|
|
|
|||
|
|
@ -81,6 +81,11 @@ check-config accepts both synthetic configurations. Real identity/custody owner
|
|||
returns, executor/recovery-state integration, exact current image verification
|
||||
and the supervised maintenance action remain open.
|
||||
|
||||
Follow-up review reproduced two collision gaps: implicit client-ID audience and
|
||||
different env names referencing the same proposed agent Secret/key. Both are now
|
||||
refused; two new regressions failed before repair and pass afterward. This does
|
||||
not establish isolation outside the observed registry/issuer environment.
|
||||
|
||||
## Verify issuer and real consumer acceptance
|
||||
|
||||
```task
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue