import importlib.util
import json
import os
import shutil
import subprocess
import sys
import time
from pathlib import Path
import yaml

os.umask(0o077)
R=Path(__file__).parent
s=importlib.util.spec_from_file_location('prior','/tmp/codex/ltb1021/2026-09-15_contextra-rtsp503_r01/probe.py')
prior=importlib.util.module_from_spec(s);s.loader.exec_module(prior)
base=prior.ops;api=prior.prev.api;REPO=base.REPO;C=base.C
PRESET=Path('/workspace/contextra-runtime-config/test-run/presets/intrusion-demo.yaml')
CLI=[sys.executable,'-B',str(REPO/'tools/test_run.py')]

def save(name,value):(R/name).write_text(json.dumps(value,ensure_ascii=False,indent=2))

def gate():
    assert not (R/'plan.json').exists()
    state=api('/api/v1/streams/rtsp-test-01');settings=api('/api/v1/streams/rtsp-test-01/settings')['configured']
    sender=prior.prev.sender();assert sender['state']=='STREAMING'
    registry=yaml.safe_load(Path('/data/contextra/runtime-state/streams_runtime.yaml').read_text())
    cfg=next(x for x in registry['added'] if x['stream_id']=='rtsp-test-01')
    assert cfg['uri']==sender['uri'] and cfg['store'] is False and cfg['active_learning'] is False
    for file in ['access.json','firewall.py','users.acl','compose.yaml','redis.conf']:shutil.copy2(C/file,R/(file+'.before'))
    rules=subprocess.check_output(['iptables','-w','5','-S','CTX_METADATA_LAN'],text=True)
    assert '192.168.50.19/32' not in rules
    save('plan.json',dict(started=time.time(),head=subprocess.check_output(['git','rev-parse','HEAD'],cwd=REPO,text=True).strip(),state=state,settings=settings,sender=sender,registration=cfg,registry_hash=base.g.sha(Path('/data/contextra/runtime-state/streams_runtime.yaml')),others=base.inventory(),redis=base.inspect('contextra-redis-1'),rules=rules,config_hashes={file:base.g.sha(C/file) for file in ['access.json','firewall.py','users.acl','compose.yaml','redis.conf']},budget_seconds=1200,caps=dict(first_frame_wait=90,stop_wait=90,helper_seconds=240,measurement_seconds=15,stored_frames=10,memory_peak_bytes=12*1024**3,min_available_ram_bytes=8*1024**3,gpu_mib=16000),rollback='Firewall: exact access/firewall backup and remove only added source rule. CLI failure: no repeated Stop; preserve first error then safe-stop Analytics only if all other streams stopped. Sender never controlled.'))
    PRESET.parent.mkdir(parents=True,exist_ok=True,mode=0o700)
    assert not PRESET.exists(),'PRESET_ALREADY_EXISTS'
    preset=dict(version=1,camera={k:cfg[k] for k in ['stream_id','server_id','source_id','uri']},settings={k:settings[k] for k in ['analysisFpsTarget','display_score','display_categories','debug_all_candidates','intrusion']},debug_metrics=False,wait_seconds=90)
    preset['settings']['frame_time_anchor']=sender['anchor']['frame_time_anchor']
    PRESET.write_text(yaml.safe_dump(preset,allow_unicode=True,sort_keys=False));PRESET.chmod(0o600)
    save('preset-prepared.json',dict(path=str(PRESET),sha256=base.g.sha(PRESET),anchor_input='Operator-equivalent one-time preparation; test CLI does not query sender'))
    save('ledger.json',dict(firewall_apply=0,cli_stop=0,cli_start=0,cli_restart=0,sender_control=0,redis_restart=0))
    print('Gate PASS; private preset prepared; no API mutation')

def firewall():
    plan=json.loads((R/'plan.json').read_text());assert not (R/'firewall-started.json').exists()
    assert base.g.sha(C/'access.json')==plan['config_hashes']['access.json']
    save('firewall-started.json',dict(at=time.time(),new_reader='192.168.50.19',rollback='Remove only new source RETURN, restore backup on exact current hash; no chain flush'))
    access=json.loads((C/'access.json').read_text());assert '192.168.50.19' not in access['readers']
    access['readers'].append('192.168.50.19')
    temporary=C/'.access-test-run.tmp';temporary.write_text(json.dumps(access,indent=2));temporary.chmod(0o600);os.replace(temporary,C/'access.json')
    try:
        p=subprocess.run([sys.executable,str(C/'firewall.py'),'apply'],capture_output=True,text=True,timeout=30);assert p.returncode==0,p.stderr
        verify=subprocess.run([sys.executable,str(C/'firewall.py'),'check'],capture_output=True,text=True,timeout=30);assert verify.returncode==0,verify.stderr
        rules=subprocess.check_output(['iptables','-w','5','-S','CTX_METADATA_LAN'],text=True)
        assert rules.index('192.168.50.19/32')<rules.index('-j DROP')
        # A second apply must not duplicate or reorder rules.
        subprocess.run([sys.executable,str(C/'firewall.py'),'apply'],capture_output=True,check=True,timeout=30)
        assert subprocess.check_output(['iptables','-w','5','-S','CTX_METADATA_LAN'],text=True)==rules
        save('firewall-result.json',dict(status='PASS',rules=rules,access_hash=base.g.sha(C/'access.json'),script_hash=base.g.sha(C/'firewall.py'),no_op=True,remote_connection='NOT_TESTED'))
    except Exception:
        subprocess.run(['iptables','-w','5','-D','CTX_METADATA_LAN','-s','192.168.50.19/32','-j','RETURN'],capture_output=True,timeout=10)
        shutil.copy2(R/'access.json.before',C/'access.json');shutil.copy2(R/'firewall.py.before',C/'firewall.py')
        raise
    ledger=json.loads((R/'ledger.json').read_text());ledger['firewall_apply']=1;save('ledger.json',ledger)
    print('PASS: exact new reader before DROP, existing scope preserved, second apply no-op')

if __name__=='__main__':{'gate':gate,'firewall':firewall}[sys.argv[1]]()