Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
100 changes: 73 additions & 27 deletions app/evidence.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@
from .models import LedgerEvent,PayloadBlob
from .service import get_events,summarize_decision,evidence_coverage
from .signing import signer
from .crypto import canonical_json,sha256_hex
from .version import VERSION,EVIDENCE_PROFILE
from .checkpoints import latest_checkpoint_covering,checkpoint_to_dict

Expand Down Expand Up @@ -72,7 +73,7 @@ def _chain_witnesses(db, events, through_seq=None):
]

def build_bundle(db:Session,decision_id:str,*,include_payloads:bool=True):
"""Build Evidence Bundle v2 while preserving the v0.6-compatible core files.
"""Build Evidence Bundle v2 with an additive signed file-attestation extension.

The bundle deliberately separates: signed ledger proof, human-readable decision
projection, signer identity, policy provenance, lifecycle projection, and optional
Expand Down Expand Up @@ -146,6 +147,7 @@ def build_bundle(db:Session,decision_id:str,*,include_payloads:bool=True):
"signer":signer.posture(),
"checkpoint":cp_json,
"files":{
"decision":"decision.json",
"signed_events":"events.jsonl",
"chain_witnesses":"chain-witness.jsonl",
"signer":"signer.json",
Expand All @@ -154,9 +156,17 @@ def build_bundle(db:Session,decision_id:str,*,include_payloads:bool=True):
"policy":"policy/policy.json" if include_payloads and policy else None,
"public_key":"public-key.pem",
"report":"report.html",
"readme":"README.txt",
"bundle_attestation":"bundle-attestation.json",
"disclosures":"disclosures.jsonl" if disclosures else None,
"timestamp":"timestamp.tsr" if cp and cp.timestamp_token_b64 else None,
},
"bundle_integrity":{
"mode":"signed_file_attestation",
"attestation_schema":"loopgrid/bundle-attestation/1",
"attestation_file":"bundle-attestation.json",
"hash_algorithm":"SHA-256",
},
"independent_verification":"python loopgrid_verify.py evidence.zip",
"cryptographic_profile":f"LoopGrid Evidence Profile {EVIDENCE_PROFILE}",
"legal_note":"Evidence support only; not a legal compliance determination.",
Expand All @@ -168,32 +178,68 @@ def build_bundle(db:Session,decision_id:str,*,include_payloads:bool=True):
report_summary=summary if include_payloads else public_summary
report_events=hydrated if include_payloads else events
report=render_report(report_summary,report_events,verification,coverage)
readme=(
'LoopGrid Evidence Bundle v2\n\n'
'Verify integrity: python loopgrid_verify.py <bundle.zip>\n'
'Pin signer identity: python loopgrid_verify.py <bundle.zip> --expected-key-id <key-id>\n'
'or: python loopgrid_verify.py <bundle.zip> --trusted-public-key <public.pem>\n\n'
'No LoopGrid server connection is required. The embedded public key proves integrity under that key; '
'signer authenticity should be pinned out-of-band for high-assurance use.\n'
'This export includes a signed SHA-256 file attestation covering manifest.json and every other exported evidence file; the attestation itself is digitally signed.\n'
'FULL mode raw payloads are encrypted outside the signed ledger; disclosures.jsonl is optional and commitment-checked.\n'
'chain-witness.jsonl contains proof-only bridge nodes and no unrelated decision payloads.\n'
'verification.json records export-time server verification; always run the offline verifier independently when relying on the evidence.\n'
)

# Build every payload as bytes before writing the ZIP so the exact exported bytes can
# be hashed and covered by a signer-authenticated bundle attestation. The attestation
# is deliberately separate from the workspace checkpoint: checkpoints seal ledger
# state, while this export-time attestation seals the portable bundle representation.
payloads={
'decision.json':json.dumps(public_summary,indent=2,ensure_ascii=False).encode('utf-8'),
'events.jsonl':'\n'.join(json.dumps(e,ensure_ascii=False) for e in events).encode('utf-8'),
'chain-witness.jsonl':'\n'.join(json.dumps(w,ensure_ascii=False) for w in witnesses).encode('utf-8'),
'signer.json':json.dumps(signer_json,indent=2,ensure_ascii=False).encode('utf-8'),
'verification.json':json.dumps(verification_json,indent=2,ensure_ascii=False).encode('utf-8'),
'lifecycle.json':json.dumps(lifecycle,indent=2,ensure_ascii=False).encode('utf-8'),
'public-key.pem':signer.public_key_pem(),
'report.html':report.encode('utf-8'),
'README.txt':readme.encode('utf-8'),
}
if include_payloads and policy:
payloads['policy/policy.json']=json.dumps(policy,indent=2,ensure_ascii=False).encode('utf-8')
if disclosures:
payloads['disclosures.jsonl']='\n'.join(json.dumps(d,ensure_ascii=False) for d in disclosures).encode('utf-8')
if cp and cp.timestamp_token_b64:
payloads['timestamp.tsr']=base64.b64decode(cp.timestamp_token_b64)

manifest_bytes=json.dumps(manifest,indent=2,ensure_ascii=False).encode('utf-8')
attested_files={'manifest.json':sha256_hex(manifest_bytes)}
attested_files.update({name:sha256_hex(data) for name,data in payloads.items()})
attestation_body={
'attestation_schema':'loopgrid/bundle-attestation/1',
'bundle_schema':manifest['bundle_schema'],
'decision_id':decision_id,
'workspace_id':summary.get('workspace_id'),
'hash_algorithm':'SHA-256',
'files':attested_files,
'signer':{
'key_id':signer.key_id,
'algorithm':signer.algorithm,
},
}
attestation_digest=sha256_hex(canonical_json(attestation_body))
attestation={
**attestation_body,
'attestation_digest':attestation_digest,
'signature':signer.sign_hash(attestation_digest),
}
attestation_bytes=json.dumps(attestation,indent=2,ensure_ascii=False).encode('utf-8')

buf=io.BytesIO()
with zipfile.ZipFile(buf,'w',zipfile.ZIP_DEFLATED) as z:
z.writestr('manifest.json',json.dumps(manifest,indent=2,ensure_ascii=False))
z.writestr('decision.json',json.dumps(public_summary,indent=2,ensure_ascii=False))
z.writestr('events.jsonl','\n'.join(json.dumps(e,ensure_ascii=False) for e in events))
z.writestr('chain-witness.jsonl','\n'.join(json.dumps(w,ensure_ascii=False) for w in witnesses))
z.writestr('signer.json',json.dumps(signer_json,indent=2,ensure_ascii=False))
z.writestr('verification.json',json.dumps(verification_json,indent=2,ensure_ascii=False))
z.writestr('lifecycle.json',json.dumps(lifecycle,indent=2,ensure_ascii=False))
if include_payloads and policy:
z.writestr('policy/policy.json',json.dumps(policy,indent=2,ensure_ascii=False))
if disclosures:
z.writestr('disclosures.jsonl','\n'.join(json.dumps(d,ensure_ascii=False) for d in disclosures))
z.writestr('public-key.pem',signer.public_key_pem())
z.writestr('report.html',report)
if cp and cp.timestamp_token_b64:
z.writestr('timestamp.tsr',base64.b64decode(cp.timestamp_token_b64))
z.writestr('README.txt',
'LoopGrid Evidence Bundle v2\n\n'
'Verify integrity: python loopgrid_verify.py <bundle.zip>\n'
'Pin signer identity: python loopgrid_verify.py <bundle.zip> --expected-key-id <key-id>\n'
'or: python loopgrid_verify.py <bundle.zip> --trusted-public-key <public.pem>\n\n'
'No LoopGrid server connection is required. The embedded public key proves integrity under that key; '\
'signer authenticity should be pinned out-of-band for high-assurance use.\n'
'FULL mode raw payloads are encrypted outside the signed ledger; disclosures.jsonl is optional and commitment-checked.\n'
'chain-witness.jsonl contains proof-only bridge nodes and no unrelated decision payloads.\n'
'verification.json records export-time server verification; always run the offline verifier independently when relying on the evidence.\n'
)
z.writestr('manifest.json',manifest_bytes)
for name,data in payloads.items():
z.writestr(name,data)
z.writestr('bundle-attestation.json',attestation_bytes)
return buf.getvalue(),f"loopgrid-evidence-{decision_id}.zip"
102 changes: 100 additions & 2 deletions tests/test_flow.py
Original file line number Diff line number Diff line change
Expand Up @@ -435,19 +435,117 @@ def test_v07_policy_digest_and_input_commitment_are_deterministic_and_version_bo
assert c['version']=='17.4' and c['policy_digest']!=a['policy_digest'] and c['input_commitment']==a['input_commitment']


def test_v07_evidence_bundle_v2_contains_trust_lifecycle_policy_and_verifies(tmp_path):
def test_evidence_bundle_v2_signed_file_attestation_verifies(tmp_path):
reset();did=client.post('/api/v1/demo/refund').json()['summary']['decision_id']
p=tmp_path/'v07-bundle.zip';p.write_bytes(client.get(f'/api/v1/decisions/{did}/evidence').content)
import zipfile,json
with zipfile.ZipFile(p) as z:
names=set(z.namelist());manifest=json.loads(z.read('manifest.json'))
assert {'signer.json','verification.json','lifecycle.json','policy/policy.json','public-key.pem','events.jsonl','chain-witness.jsonl'} <= names
assert {'signer.json','verification.json','lifecycle.json','policy/policy.json','public-key.pem','events.jsonl','chain-witness.jsonl','decision.json','report.html','README.txt','bundle-attestation.json'} <= names
assert manifest['bundle_schema']=='loopgrid/evidence-bundle/2' and manifest['version']=='3.0-draft'
assert manifest['policy']['policy_digest'] and manifest['lifecycle']['state']=='evidence_complete'
result=verify_bundle(str(p))
assert result['valid'] is True and result['bundle_schema']=='loopgrid/evidence-bundle/2' and result['policy_digest']
assert result['bundle_integrity']['status']=='attested'
assert result['bundle_integrity']['attested'] is True



def test_evidence_bundle_signed_attestation_detects_projection_and_auxiliary_file_tampering(tmp_path):
import io, zipfile

reset();did=client.post('/api/v1/demo/refund').json()['summary']['decision_id']
original=client.get(f'/api/v1/decisions/{did}/evidence').content

def rewrite(mutated_name=None,remove_name=None,extra_name=None):
src=zipfile.ZipFile(io.BytesIO(original))
out=io.BytesIO()
with src,zipfile.ZipFile(out,'w',zipfile.ZIP_DEFLATED) as dst:
for info in src.infolist():
if info.filename==remove_name:
continue
data=src.read(info.filename)
if info.filename==mutated_name:
data=data+b'\nTAMPERED'
dst.writestr(info,data)
if extra_name:
dst.writestr(extra_name,b'unattested')
return out.getvalue()

for name in ('decision.json','report.html','verification.json','lifecycle.json','README.txt'):
p=tmp_path/f"tampered-{name.replace('/','-')}"
p.write_bytes(rewrite(mutated_name=name))
result=verify_bundle(str(p))
assert result['valid'] is False
assert any(f.get('reason')=='bundle_file_digest_mismatch' and f.get('file')==name for f in result['failures'])

removed=tmp_path/'missing-report.zip';removed.write_bytes(rewrite(remove_name='report.html'))
result=verify_bundle(str(removed))
assert result['valid'] is False
assert any(f.get('reason')=='attested_file_missing' and f.get('file')=='report.html' for f in result['failures'])

extra=tmp_path/'extra-file.zip';extra.write_bytes(rewrite(extra_name='untracked.txt'))
result=verify_bundle(str(extra))
assert result['valid'] is False
assert any(f.get('reason')=='unattested_archive_file' and f.get('file')=='untracked.txt' for f in result['failures'])


def test_evidence_bundle_signed_attestation_detects_manifest_and_attestation_tampering(tmp_path):
import io, json, zipfile

reset();did=client.post('/api/v1/demo/refund').json()['summary']['decision_id']
original=client.get(f'/api/v1/decisions/{did}/evidence').content

def rewrite(name,transform):
src=zipfile.ZipFile(io.BytesIO(original));out=io.BytesIO()
with src,zipfile.ZipFile(out,'w',zipfile.ZIP_DEFLATED) as dst:
for info in src.infolist():
data=src.read(info.filename)
if info.filename==name:data=transform(data)
dst.writestr(info,data)
return out.getvalue()

manifest_zip=tmp_path/'manifest-tampered.zip'
manifest_zip.write_bytes(rewrite('manifest.json',lambda raw: raw.replace(b'"legal_note"',b'"legal_note_tampered"',1)))
result=verify_bundle(str(manifest_zip))
assert result['valid'] is False
assert any(f.get('reason')=='bundle_file_digest_mismatch' and f.get('file')=='manifest.json' for f in result['failures'])

def alter_attestation(raw):
obj=json.loads(raw);obj['decision_id']='dec_tampered'
return json.dumps(obj,indent=2).encode()
attestation_zip=tmp_path/'attestation-tampered.zip'
attestation_zip.write_bytes(rewrite('bundle-attestation.json',alter_attestation))
result=verify_bundle(str(attestation_zip))
assert result['valid'] is False
assert any(f.get('reason') in {'bundle_attestation_decision_mismatch','bundle_attestation_digest_mismatch','bundle_attestation_signature_invalid'} for f in result['failures'])


def test_legacy_unattested_bundle_v2_remains_ledger_verifiable_with_warning(tmp_path):
import io, json, zipfile

reset();did=client.post('/api/v1/demo/refund').json()['summary']['decision_id']
original=client.get(f'/api/v1/decisions/{did}/evidence').content
src=zipfile.ZipFile(io.BytesIO(original));out=io.BytesIO()
with src,zipfile.ZipFile(out,'w',zipfile.ZIP_DEFLATED) as dst:
for info in src.infolist():
if info.filename=='bundle-attestation.json':
continue
data=src.read(info.filename)
if info.filename=='manifest.json':
manifest=json.loads(data)
manifest['bundle_schema']='loopgrid/evidence-bundle/2'
manifest.pop('bundle_integrity',None)
manifest.get('files',{}).pop('bundle_attestation',None)
data=json.dumps(manifest,indent=2,ensure_ascii=False).encode('utf-8')
dst.writestr(info,data)
p=tmp_path/'legacy-v2.zip';p.write_bytes(out.getvalue())
result=verify_bundle(str(p))
assert result['valid'] is True
assert result['bundle_integrity']['status']=='legacy_unattested'
assert result['bundle_integrity']['attested'] is False
assert any(w.get('reason')=='bundle_file_attestation_unavailable' for w in result['warnings'])

def test_v07_request_body_limit_and_system_posture_are_exposed():
# Content-Length is checked before parsing the body, providing an inexpensive abuse guard.
too_big=client.get('/health',headers={'Content-Length':'3000000'})
Expand Down
1 change: 1 addition & 0 deletions tests/test_otlp_http.py
Original file line number Diff line number Diff line change
Expand Up @@ -258,4 +258,5 @@ def test_protobuf_ingested_evidence_remains_v1_verifier_compatible(tmp_path) ->
verified = verify_bundle(str(bundle))
assert verified["valid"] is True
assert verified["bundle_schema"] == "loopgrid/evidence-bundle/2"
assert verified["bundle_integrity"]["attested"] is True
assert verified["signature_algorithm"] == "Ed25519"
Loading
Loading