tegwick
fb7d56d3dc
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a137ca-5c77-7a71-b9ae-23ad9b0b5684
94 lines
6.7 KiB
Python
94 lines
6.7 KiB
Python
"""Local chart and non-disclosure checks only; never contacts a cluster."""
|
|
import hashlib
|
|
import importlib.util
|
|
import json
|
|
from pathlib import Path
|
|
import subprocess
|
|
import yaml
|
|
|
|
ROOT = Path(__file__).resolve().parents[3]
|
|
ARTIFACT = Path(__file__).resolve().parent
|
|
calls = []
|
|
|
|
def render(extra, success=True):
|
|
command = ['helm', 'template', 'reuse', str(ROOT / 'charts/reuse-surface'), '-f', str(ROOT / 'helm/reuse-surface-values.yaml'), *extra]
|
|
proc = subprocess.run(command, capture_output=True, text=True)
|
|
calls.append({'command': command, 'exit_code': proc.returncode, 'expected_success': success})
|
|
assert (proc.returncode == 0) == success
|
|
return list(yaml.safe_load_all(proc.stdout)) if success else []
|
|
|
|
def api(docs):
|
|
return next(d for d in docs if d and d.get('kind') == 'Deployment' and d['metadata']['name'] == 'reuse-surface')
|
|
|
|
current = api(render([]))
|
|
assert current['spec']['template']['spec']['containers'][0]['image'].endswith(':main-6cbc862')
|
|
assert next(e['value'] for e in current['spec']['template']['spec']['containers'][0]['env'] if e['name']=='REUSE_SURFACE_AUTH_SOURCE_ALLOWLIST') == '{}'
|
|
proposal = ['-f', str(ARTIFACT / 'private-reader-values.r1.yaml')]
|
|
render(proposal, False) # unknown real digest is a hard gate
|
|
# Synthetic digest is render-only, and expressly not promotion/build evidence.
|
|
render(proposal+['--set', 'image.digest=sha256:bad'], False)
|
|
render(proposal+['--set-json', 'authSourceAllowlist={"BAD":["http://attacker.invalid"]}', '--set', 'image.digest=sha256:'+'0'*64], False)
|
|
docs = render(proposal+['--set', 'image.digest=sha256:'+'0'*64])
|
|
deployment = api(docs)
|
|
container = deployment['spec']['template']['spec']['containers'][0]
|
|
assert container['image'].endswith('@sha256:'+'0'*64)
|
|
allowlist=json.loads(next(e['value'] for e in container['env'] if e['name']=='REUSE_SURFACE_AUTH_SOURCE_ALLOWLIST'))
|
|
expected = json.loads((ARTIFACT/'runtime-auth-source-allowlist-r1.json').read_text())
|
|
assert allowlist==expected and len(next(iter(allowlist.values())))==14
|
|
assert deployment['spec']['strategy'] == current['spec']['strategy']
|
|
assert container['resources']==current['spec']['template']['spec']['containers'][0]['resources']
|
|
assert deployment['spec']['template']['spec']['volumes']==current['spec']['template']['spec']['volumes']
|
|
assert container['volumeMounts']==current['spec']['template']['spec']['containers'][0]['volumeMounts']
|
|
secret=yaml.safe_load((ARTIFACT/'reuse-surface-runtime-externalsecret.r1.yaml').read_text())
|
|
baseline=yaml.safe_load((ROOT/'manifests/reuse-surface-runtime-externalsecret.yaml').read_text())
|
|
assert secret['spec']['data'][:-1]==baseline['spec']['data']
|
|
assert secret['spec']['secretStoreRef']==baseline['spec']['secretStoreRef']
|
|
assert secret['spec']['data'][-1]=={'secretKey':'REUSE_SURFACE_FORGEJO_SOURCE_READ_TOKEN','remoteRef':{'key':'workloads/reuse/reuse-surface/forgejo-source-read','property':'FORGEJO_SOURCE_TOKEN'}}
|
|
spec=importlib.util.spec_from_file_location('inventory',ROOT/'scripts/reuse_authenticated_source_inventory.py')
|
|
mod=importlib.util.module_from_spec(spec);spec.loader.exec_module(mod)
|
|
url=expected['REUSE_SURFACE_FORGEJO_SOURCE_READ_TOKEN'][1]
|
|
assert len(mod.sanitize([['rapp-openbao','FIXTURE_READ_TOKEN',url]]))==1
|
|
legacy='https://forgejo.coulomb.social/coulomb/rapp-openbao/raw/branch/main/registry/indexes/capabilities.yaml'
|
|
assert len(mod.sanitize([['rapp-openbao','FIXTURE_READ_TOKEN',legacy]]))==1
|
|
# Exercise the EXACT pod code against actual native registration keys in SQLite.
|
|
# Replacing only the fixture DB path leaves the metadata SELECT/read-only mode intact.
|
|
import sqlite3,tempfile
|
|
with tempfile.TemporaryDirectory(prefix='apps-reader-fixture-') as temp:
|
|
fixture=Path(temp)/'reuse.db'
|
|
with sqlite3.connect(fixture) as db:
|
|
db.execute('CREATE TABLE registrations(repo TEXT PRIMARY KEY, payload TEXT NOT NULL, registered_at TEXT, updated_at TEXT)')
|
|
db.execute('INSERT INTO registrations VALUES(?,?,?,?)', ('rapp-openbao',json.dumps({'repo':'rapp-openbao','url':url,'auth_env':'FIXTURE_READ_TOKEN','description':'fixture-secret-never-emit'}),'fixture','fixture'))
|
|
db.execute('INSERT INTO registrations VALUES(?,?,?,?)', ('anonymous',json.dumps({'repo':'anonymous','url':legacy}),'fixture','fixture'))
|
|
pod_code=mod.POD_CODE.replace('/data/reuse.db',str(fixture))
|
|
proc=subprocess.run(['python3','-c',pod_code],capture_output=True,text=True,check=True)
|
|
assert json.loads(proc.stdout)==[['rapp-openbao','FIXTURE_READ_TOKEN',url]]
|
|
assert 'fixture-secret' not in proc.stdout
|
|
assert len(mod.sanitize(json.loads(proc.stdout)))==1
|
|
with sqlite3.connect(fixture) as db: assert db.execute('SELECT count(*) FROM registrations').fetchone()==(2,)
|
|
assert 'mode=ro' in pod_code and '$.index_url' not in pod_code
|
|
assert 'os.environ' not in pod_code and 'getenv' not in pod_code and 'kubectl' not in pod_code
|
|
|
|
for bad in ['https://token@forgejo.coulomb.social/x', url+'&token=fixture-secret','http://attacker.invalid',url+'#fixture-secret',legacy+'?token=fixture-secret',legacy+'/../credentials',legacy+'%3Ftoken=fixture-secret']:
|
|
try: mod.sanitize([['rapp-openbao','FIXTURE_READ_TOKEN',bad]])
|
|
except ValueError: pass
|
|
else: raise AssertionError('unsafe URL accepted')
|
|
# Errors from kubectl are discarded before output; no subprocess invoked in test.
|
|
original=mod.subprocess.run
|
|
class Failure:
|
|
returncode=1;stdout=b'fixture-secret';stderr=b'fixture-secret'
|
|
mod.subprocess.run=lambda *args,**kwargs: Failure()
|
|
import sys,io,contextlib
|
|
argv=sys.argv;sys.argv=['inventory','--kubeconfig','fixture-path']
|
|
buf=io.StringIO()
|
|
with contextlib.redirect_stdout(buf): assert mod.main()==1
|
|
assert 'fixture-secret' not in buf.getvalue()
|
|
class UnsafeSuccess:
|
|
returncode=0;stdout=json.dumps([['rapp-openbao','FIXTURE_READ_TOKEN',url+'&token=fixture-secret']]).encode();stderr=b''
|
|
mod.subprocess.run=lambda *args,**kwargs: UnsafeSuccess()
|
|
buf=io.StringIO()
|
|
with contextlib.redirect_stdout(buf): assert mod.main()==1
|
|
assert 'fixture-secret' not in buf.getvalue()
|
|
sys.argv=argv;mod.subprocess.run=original
|
|
result={'status':'passed','local_only':True,'real_image_digest':None,'synthetic_digest_used_only_to_check_rendering':True,'checks':['production rendering unchanged except empty fail-closed allowlist','unknown/malformed digest fails','HTTP allowlist fails schema','exact14 URLs roundtrip','strategy resources PVC retained','dedicated ESO mapping preserves current fields/store','metadata sanitizer excludes userinfo/query secrets/fragments','kubectl failure output discarded','exact pod SQL real payload URL keys readonly fixture and anonymous filtering','legacy Forgejo branch URL accepted without weakening credential query refusal'],'helm_calls':calls}
|
|
(ARTIFACT/'local-verification-r2.json').write_text(json.dumps(result,indent=2)+'\n')
|
|
print(json.dumps({'status':'passed','checks':len(result['checks'])}))
|