tegwick
4f65e41ac5
Some checks failed
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a1468b-63fa-7710-85dc-8f2e16b3ebfa
98 lines
7.5 KiB
Python
98 lines
7.5 KiB
Python
"""Generated archive/key only; actual age authentication + archive restore.
|
|
|
|
Docker and issuer behavior are intentionally mocked here, not native evidence.
|
|
"""
|
|
import base64
|
|
from datetime import datetime,timedelta,timezone
|
|
import hashlib
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import subprocess
|
|
import time
|
|
from unittest.mock import MagicMock,patch
|
|
import pytest
|
|
from cryptography import x509
|
|
from cryptography.hazmat.primitives import hashes,serialization
|
|
from cryptography.hazmat.primitives.asymmetric import rsa
|
|
from cryptography.x509.oid import NameOID
|
|
from test_identity_bundle import stage,b
|
|
|
|
ROOT=Path(__file__).parents[1]
|
|
def load(name,path):
|
|
spec=importlib.util.spec_from_file_location(name,path);m=importlib.util.module_from_spec(spec);spec.loader.exec_module(m);return m
|
|
s=load('session_manifest',ROOT/'tools/bootstrap-recovery/isolated-keycape-session.py')
|
|
h=load('helper_manifest',ROOT/'sso-mfa/bootstrap/sops-custody-fd.py')
|
|
|
|
@pytest.fixture
|
|
def envelope(stage,tmp_path,monkeypatch):
|
|
raw=subprocess.run(['age-keygen'],stdout=subprocess.PIPE,stderr=subprocess.DEVNULL,check=True).stdout
|
|
key=h.sealed_key(next(line for line in raw.splitlines() if line.startswith(b'AGE-SECRET-KEY-'))+b'\n')
|
|
recipient=subprocess.run(['age-keygen','-y',f'/proc/self/fd/{key}'],pass_fds=(key,),stdout=subprocess.PIPE,stderr=subprocess.DEVNULL,check=True).stdout.decode().strip()
|
|
profile=b.profile();profile['public_recipient']=recipient;monkeypatch.setattr(b,'profile',lambda:profile);monkeypatch.setattr(s.bundle,'profile',lambda:profile)
|
|
artifact=tmp_path/'identity.tar.age';sealed=b.seal(stage,artifact,{'test_only':True})
|
|
signer=rsa.generate_private_key(public_exponent=65537,key_size=2048);name=x509.Name([x509.NameAttribute(NameOID.COMMON_NAME,'synthetic.invalid')]);now=datetime.now(timezone.utc)
|
|
cert=(x509.CertificateBuilder().subject_name(name).issuer_name(name).public_key(signer.public_key()).serial_number(1).not_valid_before(now-timedelta(minutes=1)).not_valid_after(now+timedelta(hours=1)).sign(signer,hashes.SHA256())).public_bytes(serialization.Encoding.PEM)
|
|
tls={'/recovery-tls/server.pem':cert,'/recovery-tls/trust.pem':cert,'/recovery-tls/server-key.pem':signer.private_bytes(serialization.Encoding.PEM,serialization.PrivateFormat.TraditionalOpenSSL,serialization.NoEncryption())}
|
|
transfer=dict(schema='netkingdom.identity-offsite.v1',status='readback_verified',provider='Storage Share',folder='Backup/railiance-backups/net-kingdom-identity',artifact='net-kingdom-identity-20261016T063159Z-5882771aa2ce.tar.age',etag='"synthetic"',ciphertext_sha256=sealed['ciphertext_sha256'],ciphertext_bytes=artifact.stat().st_size,fetched_path=str(artifact),independent_owner_retrieval_verified=True)
|
|
admission=dict(ciphertext_sha256=sealed['ciphertext_sha256'],manifest_binding_mode='ciphertext-derived-v1',issuer='https://synthetic.invalid',tls_cert_sha256=hashlib.sha256(cert).hexdigest(),trust_sha256=hashlib.sha256(cert).hexdigest(),verifier_sha256=hashlib.sha256(b'synthetic executable').hexdigest())
|
|
try:yield artifact,key,transfer,admission,tls
|
|
finally:os.close(key)
|
|
|
|
def execute(envelope,*,wrong_key=False):
|
|
artifact,key,transfer,admission,tls=envelope;receipt={};seen={};checkpoints=[]
|
|
if wrong_key:
|
|
raw=subprocess.run(['age-keygen'],stdout=subprocess.PIPE,stderr=subprocess.DEVNULL,check=True).stdout
|
|
key=h.sealed_key(next(line for line in raw.splitlines() if line.startswith(b'AGE-SECRET-KEY-'))+b'\n')
|
|
actualrun=s.run
|
|
def run(argv,**kwargs):return actualrun(argv,**kwargs) if argv[0]=='age-keygen' else b''
|
|
def native(root,manifest,checked,receipt,**kwargs):seen.update(checked);receipt['status']='synthetic_flow_checked'
|
|
execution={'pid':os.getpid(),'sid':os.getsid(0),'pgid':os.getpgrp(),'uid':os.getuid(),'deadline':time.monotonic()+60}
|
|
try:
|
|
with patch.object(s,'verify_execution'),patch.object(s.local,'preflight',return_value={}),patch.object(s.local.verify.custody,'private_key_fd',side_effect=lambda:os.dup(key)),patch.object(s,'start_reaper',return_value=MagicMock()),patch.object(s,'run',side_effect=run),patch.object(s.adapter,'container_file',return_value=b'synthetic executable'),patch.object(s.adapter,'private_payload',return_value={'files':{'/etc/keycape/config.yaml':base64.b64encode(b'issuer: https://synthetic.invalid\nport: 8080').decode()},'env':{}}),patch.object(s,'provision'),patch.object(s.adapter,'check_jwks'),patch.object(s.adapter,'native_checks',side_effect=native),patch.object(s,'cleanup_all'):
|
|
# Key file must be valid at the later public signer construction.
|
|
payload_key=rsa.generate_private_key(public_exponent=65537,key_size=2048).private_bytes(serialization.Encoding.PEM,serialization.PrivateFormat.TraditionalOpenSSL,serialization.NoEncryption())
|
|
with patch.object(s.adapter,'private_payload',return_value={'files':{'/etc/keycape/config.yaml':base64.b64encode(b'issuer: https://synthetic.invalid\nport: 8080').decode(),'/etc/keycape/key.pem':base64.b64encode(payload_key).decode()},'env':{}}):
|
|
s.execute(transfer,admission,tls,receipt,execution,checkpoint=lambda r:checkpoints.append(json.loads(json.dumps(r))))
|
|
return receipt,seen,checkpoints
|
|
finally:
|
|
if wrong_key:os.close(key)
|
|
|
|
def test_exact_ciphertext_authenticated_restore_derives_private_manifest_without_safe_leak(envelope):
|
|
receipt,seen,checkpoints=execute(envelope)
|
|
derived=seen['manifest_sha256'];assert len(derived)==64
|
|
safe=json.dumps([receipt,checkpoints]);assert derived not in safe;assert 'manifest_sha256' not in safe
|
|
assert receipt['all_manifest_hashes_verified'] is True
|
|
assert receipt['temporary_plaintext_cleanup_verified'] is True
|
|
|
|
def test_changed_ciphertext_refuses_before_archive_decryption(envelope):
|
|
artifact,*_=envelope;artifact.write_bytes(artifact.read_bytes()[:-1]+b'!')
|
|
with patch.object(s.adapter,'process') as process,pytest.raises(s.adapter.Refused):execute(envelope)
|
|
assert not process.called
|
|
|
|
def test_modified_ciphertext_rebound_metadata_still_fails_authenticated_age(envelope):
|
|
artifact,key,transfer,admission,tls=envelope
|
|
raw=artifact.read_bytes();artifact.write_bytes(raw[:-32]+b'!'*32)
|
|
digest=hashlib.sha256(artifact.read_bytes()).hexdigest()
|
|
transfer['ciphertext_sha256']=digest;admission['ciphertext_sha256']=digest
|
|
with pytest.raises(s.adapter.Refused):execute(envelope)
|
|
|
|
def test_wrong_key_refuses_before_decryption_and_never_exposes_manifest(envelope):
|
|
with pytest.raises(s.adapter.Refused):execute(envelope,wrong_key=True)
|
|
|
|
def test_wrong_explicit_manifest_refuses_after_full_restore(envelope):
|
|
admission=envelope[3];admission['manifest_binding_mode']='explicit-digest-v1';admission['manifest_sha256']='0'*64
|
|
with pytest.raises(s.adapter.Refused):execute(envelope)
|
|
|
|
@pytest.mark.parametrize('mode',['derived','ciphertext-derived-v2',None,True,{}])
|
|
def test_unknown_binding_mode_refuses_before_effects(envelope,mode):
|
|
envelope[3]['manifest_binding_mode']=mode
|
|
with patch.object(s.local,'preflight') as preflight,pytest.raises(s.adapter.Refused):execute(envelope)
|
|
assert not preflight.called
|
|
|
|
def test_conflicting_binding_and_unverified_transfer_refuse(envelope):
|
|
admission=envelope[3];transfer=envelope[2];admission['manifest_sha256']='0'*64
|
|
with pytest.raises(s.adapter.Refused):s.adapter.manifest_binding_mode(admission,transfer)
|
|
admission.pop('manifest_sha256');transfer['independent_owner_retrieval_verified']=False
|
|
with pytest.raises(s.adapter.Refused):s.adapter.manifest_binding_mode(admission,transfer)
|