from copy import deepcopy from datetime import datetime, timezone import importlib import json from pathlib import Path import sys import time import pytest from fastapi import FastAPI, Request from fastapi.responses import JSONResponse from fastapi.testclient import TestClient from contextra_trace.sanitize import copy_safe, MASK from contextra_trace.writer import DEFAULT, Recorder, policy from contextra_trace.http import Capture, REQUEST_ID from app.trace_store import filters @pytest.mark.parametrize('name',['password','passwd','Authorization','accessToken','api_key','Cookie','privateKey']) def test_credentials_removed_and_input_not_mutated(name): d={'cameraId':'00000003','roiId':501,'config':{'unexpected':True},name:'DO_NOT_STORE'} before=deepcopy(d);clean,marks=copy_safe(d) assert clean[name]==MASK and marks assert clean['cameraId']=='00000003' and clean['config']==before['config'] and d==before assert 'DO_NOT_STORE' not in json.dumps(clean) @pytest.mark.parametrize('raw',[b'{',b'{"password":"DO_NOT_STORE",',b'{"a":1,"a":2}',b'NaN',b'x'*65537]) def test_invalid_or_large_json_never_persists_raw(raw): clean,_=copy_safe({'requestBody':raw}) assert 'omitted' in clean['requestBody'] assert 'DO_NOT_STORE' not in json.dumps(clean) def test_url_secrets_masked_address_and_ordinary_values_preserved(): raw={'uri':'rtsp://operator:DO_NOT_STORE@camera.invalid:554/live?token=SECOND_SECRET&quality=high', 'message':'Authorization: Bearer THIRD_SECRET','enabled':'wrong-type'} clean,_=copy_safe(raw);text=json.dumps(clean) assert not any(x in text for x in ('DO_NOT_STORE','SECOND_SECRET','THIRD_SECRET')) assert 'camera.invalid:554/live' in text and 'quality=high' in text assert clean['enabled']=='wrong-type' def test_queue_full_db_failure_shutdown_are_bounded(): r=Recorder(lambda rows: (_ for _ in ()).throw(RuntimeError('secret')),server='fixture', settings=dict(DEFAULT,queue_limit=16),start=False) start=time.monotonic() for _ in range(1000):r.emit({'phase':'TEST'}) assert r.queue.qsize()==16 and r.counts['dropped_full']==984 r.flush([r.queue.get()]);assert r.error=='TRACE_SINK_UNAVAILABLE' assert time.monotonic()-start<2 r.close();assert not r.emit({'phase':'AFTER_STOP'}) def test_policy_manual_retention_and_safe_reload(tmp_path,monkeypatch): from types import SimpleNamespace monkeypatch.setattr('contextra_trace.writer.shutil.disk_usage',lambda _:SimpleNamespace(free=20*1024**3)) p=tmp_path/'policy.json';p.write_text(json.dumps(DEFAULT)) assert policy(p)['auto_cleanup'] is False r=Recorder(lambda rows:None,server='fixture',path=str(p),start=False) p.write_text(json.dumps(dict(DEFAULT,enabled=False))) with pytest.raises(ValueError,match='TRACE_DISABLED'):r.check() assert not r.emit({'phase':'TEST'}) p.write_text(json.dumps(dict(DEFAULT,enabled=True)));r.last_check=0;r.check() assert r.emit({'phase':'TEST'}) p.write_text(json.dumps(dict(DEFAULT,auto_cleanup=True))) with pytest.raises(ValueError):policy(p) p.unlink();p.symlink_to(tmp_path/'missing') with pytest.raises(ValueError):policy(p) @pytest.mark.parametrize('mode',['legacy','multi_roi']) def test_capture_preserves_raw_body_response_and_does_not_wait_for_database(mode): recorder=Recorder(lambda rows:None,server='fixture',start=False) app=FastAPI() @app.post('/api/v1/cameras/start') async def route(request:Request): assert REQUEST_ID.get() return JSONResponse({'body':(await request.body()).decode(),'result':'FAIL','code':'INVALID_REQUEST'}) app.add_middleware(Capture,recorder=recorder,component='sidecar') raw=json.dumps({'mode':mode,'cameraId':3,'rois':[],'password':'DO_NOT_STORE'},indent=2).encode() with TestClient(app) as c: response=c.post('/api/v1/cameras/start',content=raw) assert response.json()['body']==raw.decode() and response.json()['code']=='INVALID_REQUEST' d=recorder.queue.get() assert d['requestBody']['password']==MASK and d['cameraId']=='3' assert d['error']=='INVALID_REQUEST' and d['requestId'] # A quoted JSON string echoed inside an error body must not leak credentials either. assert 'DO_NOT_STORE' not in json.dumps(d) def test_sink_exception_cannot_change_response(): class Broken: def emit(self,_):raise RuntimeError('failure') app=FastAPI() @app.post('/api/v1/cameras/stop') async def stop():return {'result':'SUCCESS'} app.add_middleware(Capture,recorder=Broken(),component='sidecar') assert TestClient(app).post('/api/v1/cameras/stop',json={'cameraId':3}).json()=={'result':'SUCCESS'} def test_kst_range_and_filter_injection(): terms,args=filters({'camera':"3' OR 1=1 --",'from':'2026-10-20','to':'2026-10-21'}) assert terms[0]=='camera_id=%s' and args[0]=="3' OR 1=1 --" assert args[1].astimezone(timezone.utc).isoformat()=='2026-10-19T15:00:00+00:00' assert args[2].astimezone(timezone.utc).isoformat()=='2026-10-21T15:00:00+00:00' with pytest.raises(ValueError):filters({'from':'2026-10-22','to':'2026-10-21'}) def test_internal_trace_api_auth_server_and_db_failure(tmp_path,monkeypatch): import app.trace_runtime as runtime p=tmp_path/'policy.json';p.write_text(json.dumps(DEFAULT)) monkeypatch.setenv('CONTEXTRA_TRACE_POLICY_FILE',str(p)) monkeypatch.setenv('CONTEXTRA_TRACE_SERVER','fixture') class Store: def __init__(self,*args):pass def write(self,rows):raise RuntimeError('DB secret') def query(self,*args):raise RuntimeError('DB secret') monkeypatch.setattr(runtime,'Store',Store) app=FastAPI();runtime.install(app,'test-only-control','AI-fixture') with TestClient(app) as c: assert c.post('/internal/trace/status',json={}).status_code==401 h={'Authorization':'Bearer test-only-control'} assert c.post('/internal/trace/status',headers=h,json={'server':'wrong'}).status_code==422 assert c.post('/internal/trace/list',headers=h,json={'server':'fixture'}).json()['error']=='TRACE_DB_UNAVAILABLE' result=c.post('/internal/trace/status',headers=h,json={'server':'fixture'}).json() assert result['database']=='TRACE_DB_UNAVAILABLE' and 'DB secret' not in json.dumps(result) assert not app.state.trace.thread.is_alive() def trace_cli(monkeypatch): root=Path(__file__).resolve().parents[1]/'tools/icerti_offline/maintenance' monkeypatch.syspath_prepend(str(root)) return importlib.import_module('field_trace') def test_export_cleanup_protects_symlink_extra_file_and_modified_content(tmp_path,monkeypatch): m=trace_cli(monkeypatch);folder=tmp_path/'exports/fixture/2026-10/folder';folder.mkdir(parents=True) for name in m.FILES-{'manifest.json'}:(folder/name).write_text('fixture') manifest={'version':1,'server':'fixture','complete':True,'exportedAt':'2026-10-21T12:00:00+09:00', 'filters':{'camera':'3'},'files':{name:m.sha(folder/name) for name in m.FILES-{'manifest.json'}}} (folder/'manifest.json').write_text(json.dumps(manifest)) meta={'exports':str(tmp_path/'exports'),'server':'fixture'} rows=m.export_candidates(meta,{'camera':'3'}) (folder/'foreign-file').write_text('not owned') with pytest.raises(ValueError):m.clean_exports(rows) (folder/'foreign-file').unlink() (folder/'timeline.csv').unlink();(folder/'timeline.csv').symlink_to(tmp_path/'outside') with pytest.raises(ValueError):m.clean_exports(rows) (folder/'timeline.csv').unlink();(folder/'timeline.csv').write_text('fixture') m.clean_exports(rows);assert not folder.exists() def test_export_pagination_and_csv_formula_guard(tmp_path,monkeypatch): from types import SimpleNamespace m=trace_cli(monkeypatch);monkeypatch.setattr(m,'owner_tree',lambda _:None) monkeypatch.setattr(m.shutil,'disk_usage',lambda _:SimpleNamespace(free=20*1024**3)) monkeypatch.setattr(m.os,'chown',lambda *args:None) monkeypatch.setattr(m.server,'operator',lambda:'root') root=tmp_path/'exports';root.mkdir() calls=[] def call(container,action,body): calls.append((action,body)) if action=='plan':return {'cutoff':1,'count':1} if body['after']==0:return {'rows':[{'seq':1,'doc':{'cameraId':'=HYPERLINK("bad")'}}],'next':1} return {'rows':[],'next':1} monkeypatch.setattr(m,'call',call) dest=m.export({'exports':str(root),'server':'fixture'},'fake',{'server':'fixture','filters':{}}) assert json.loads((dest/'manifest.json').read_text())['complete'] assert "'=HYPERLINK" in (dest/'timeline.csv').read_text() assert all(body.get('cutoff')==1 for action,body in calls if action=='list') def test_event_attempt_identity_and_business_retry_unchanged(monkeypatch): from types import SimpleNamespace import contextra_trace from app.trace_events import trace_event from test_secure_bridge import artifact_module,event_body module=artifact_module(monkeypatch) recorder=Recorder(lambda rows:None,server='fixture',start=False) monkeypatch.setattr(contextra_trace,'_recorder',recorder) svc=module.ArtifactService(SimpleNamespace(public_base='https://ai.invalid',endpoint='unused')) class Store: attempts=0 def prepare_delivery(self,row,body):return body def begin_attempt(self,row): if self.attempts==2:return None self.attempts+=1;return self.attempts def finish_attempt(self,row,state,*args):row['state']=state svc.store=Store() def send(endpoint,body): trace_event('EVENT_HTTP_RESPONSE',body,httpStatus=503,responseBody=b'{"token":"DO_NOT_STORE"}') return False,True,503,None,'HTTP_REJECTED' monkeypatch.setattr(module,'post_event',send) row={'body':event_body()};svc._deliver(row);svc._deliver(row);svc._deliver(row) rows=[] while not recorder.queue.empty():rows.append(recorder.queue.get()) assert len(rows)==6 and {d['eventId'] for d in rows}=={'test-event'} assert {d['attempt'] for d in rows}=={1,2} and row['state']=='FAILED' assert 'DO_NOT_STORE' not in json.dumps(rows)