#!/usr/bin/env python3
# /// script
# requires-python = ">=3.11"
# dependencies = ["c2pa-python==0.37.12", "cryptography>=46,<47"]
# ///
"""Free local verifier. No audio upload, account, product licence or network verification.
Run: uv run verify-impact-audio.py YourSound.wav
A valid local signature proves integrity, not publisher identity or copyright clearance.
"""
import argparse, base64, hashlib, json, struct
from pathlib import Path
from cryptography import x509
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.asymmetric import ec
SCHEMA = 'https://useimpactstudio.com/provenance/v1.json'
ASSERTION = 'org.useimpactstudio.export'
AIFF_PREFIX = b'Impact Studio provenance v1\n'
AI_ORIGINS = {'ai_generated','ai_manipulated','ai_generated_or_manipulated'}
MIMES={'.wav':'audio/wav','.flac':'audio/flac','.mp3':'audio/mpeg'}
def media_kind(path):
    with Path(path).open('rb') as f: header=f.read(12)
    if header[:4]==b'RIFF' and header[8:]==b'WAVE': return '.wav'
    if header[:4]==b'FORM' and header[8:] in (b'AIFF',b'AIFC'): return '.aiff'
    if header[:4]==b'fLaC': return '.flac'
    if header[:3]==b'ID3' or (header and header[0]==255): return '.mp3'
    return Path(path).suffix.lower()

def canonical(value):
    return json.dumps(value, sort_keys=True, separators=(',', ':'), ensure_ascii=True, allow_nan=False).encode()

def _aiff_parts(data):
    if len(data)<12 or data[:4]!=b'FORM' or data[8:12] not in (b'AIFF',b'AIFC'): raise ValueError('Not an AIFF file.')
    if struct.unpack('>I',data[4:8])[0]+8 != len(data): raise ValueError('Invalid AIFF length.')
    chunks=[]; record=None; offset=12
    while offset<len(data):
        if offset+8>len(data): raise ValueError('Truncated AIFF chunk.')
        size=struct.unpack('>I',data[offset+4:offset+8])[0]; end=offset+8+size+(size%2)
        if end>len(data): raise ValueError('Truncated AIFF payload.')
        payload=data[offset+8:offset+8+size]
        if data[offset:offset+4]==b'ANNO' and payload.startswith(AIFF_PREFIX):
            if record is not None: raise ValueError('Duplicate provenance records.')
            record=json.loads(payload[len(AIFF_PREFIX):])
        else: chunks.append(data[offset:end])
        offset=end
    body=data[8:12]+b''.join(chunks)
    return b'FORM'+struct.pack('>I',len(body))+body,record

def verify(path):
    """Inspect a local file, without uploading audio or requiring a product licence."""
    path=Path(path)
    try:
        kind=media_kind(path)
        if kind in ('.aif','.aiff'):
            raw,envelope=_aiff_parts(path.read_bytes())
            if envelope is None: return dict(integrity='no_mark', origin='undetermined')
            record=envelope['record'];cert=x509.load_pem_x509_certificate(envelope['certificate'].encode())
            cert.public_key().verify(base64.b64decode(envelope['signature'],validate=True),canonical(record),ec.ECDSA(hashes.SHA256()))
            if record['media_sha256']!=hashlib.sha256(raw).hexdigest(): return dict(integrity='invalid',origin='undetermined')
            return dict(integrity='valid',origin=record['origin'],record=record,identity='locally_issued',format='signed-aiff-annotation-v1')
        import c2pa
        with c2pa.Settings.from_dict({'verify':{'remote_manifest_fetch':False,'fetch_ocsp':False}}) as settings, c2pa.Context(settings=settings) as context, path.open('rb') as stream,c2pa.Reader(MIMES.get(kind,kind),stream,context=context) as reader:
            manifest=reader.get_active_manifest() or {};record=next((a['data'] for a in manifest.get('assertions',[]) if a['label']==ASSERTION),None)
            state=reader.get_validation_state();results=reader.get_validation_results() or {}
            # A valid locally signed claim is expected to be Valid, not Trusted.
            valid=str(state).lower() in ('valid','trusted')
            if record is not None and (not isinstance(record,dict) or record.get('schema')!=SCHEMA): valid=False
            external_ai=any('trainedAlgorithmicMedia' in json.dumps(a) or 'TrainedAlgorithmicMedia' in json.dumps(a) for a in manifest.get('assertions',[]) if a['label'].startswith('c2pa.actions'))
            return dict(integrity='valid' if valid else 'invalid',origin=record.get('origin','undetermined') if record else 'ai_generated_or_manipulated' if external_ai else 'undetermined',
                        record=record,identity='locally_issued' if record else 'external',format='c2pa',validation_state=state,
                        validation=results)
    except Exception as exc:
        # Do not describe a missing/stripped mark as proof of human authorship.
        message=str(exc)
        if type(exc).__name__=='_C2paManifestNotFound' or 'JumbfNotFound' in message or 'no claim' in message.lower():
            return dict(integrity='no_mark',origin='undetermined')
        return dict(integrity='invalid',origin='undetermined',message='The file or its provenance could not be validated.')
if __name__ == '__main__':
    parser=argparse.ArgumentParser(description=__doc__)
    parser.add_argument('audio',type=Path);parser.add_argument('--json',action='store_true')
    args=parser.parse_args();result=verify(args.audio)
    if args.json: print(json.dumps(result,indent=2))
    else:
        print('Integrity:',result['integrity'])
        print('Recorded origin:',result['origin'])
        print('No supported mark is not evidence of human authorship.')
        print('A valid local signature is not copyright clearance or verified publisher identity.')
        if result.get('record'): print(json.dumps(result['record'],indent=2))
    raise SystemExit(0 if result['integrity']=='valid' else 2 if result['integrity']=='no_mark' else 1)