"""Offline TLS and loopback-only Nginx checks; never contacts deployed services.""" from contextlib import contextmanager from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path import re import shutil import socket import ssl import subprocess import threading import time import urllib.error import urllib.request import pytest import yaml from tools.secure_bridge_tls import command, prepare, verify, TLSError from control_sidecar.config import Config # Non-deployable sentinels, not credentials for a running service. CONTROL = 'control-test-token-00000000' RELAY = 'relay-test-token-0000000000' MANAGER = 'manager-test-token-000000' ROOT=Path(__file__).resolve().parents[1] @pytest.mark.parametrize('scenario', ['INTRUSION','LOITERING','FIRE','FALL','FIGHT','CROWD']) @pytest.mark.parametrize('filename', ['clip.mp4','thumbnail.jpg']) def test_gateway_media_whitelist_all_scenarios(scenario, filename): template=(ROOT/'docker/secure-gateway.conf.template').read_text() pattern=re.search(r'location ~ "([^"]+)"',template).group(1) path=f'/artifacts/cameras/3/2026-09-28/AI_153100.472+0900_{scenario}_test-event/{filename}' assert re.fullmatch(pattern,path) @pytest.mark.parametrize('relative', [ 'AI_153100.472+0900_CROWD_test-event/event.json', 'AI_153100.472+0900_CROWD_test-event/capture.bin', 'AI_153100.472+0900_CROWD_test-event/../clip.mp4', 'AI_153100.472+0900_CROWD_test-event/clip.mp4/extra', 'AI_153100.472+0900_UNKNOWN_test-event/clip.mp4', ]) def test_gateway_media_whitelist_rejects_non_media(relative): template=(ROOT/'docker/secure-gateway.conf.template').read_text() pattern=re.search(r'location ~ "([^"]+)"',template).group(1) assert not re.fullmatch(pattern,'/artifacts/cameras/3/2026-09-28/'+relative) @pytest.fixture(scope='module') def tls(tmp_path_factory): if not shutil.which('openssl'):pytest.skip('OpenSSL not installed') root=tmp_path_factory.mktemp('bridge-pki') ca,key,cert=root/'manager-ca.crt',root/'manager-ca.key',root/'control-tls.crt' command('req','-x509','-newkey','rsa:2048','-nodes','-keyout',str(key),'-out',str(ca), '-days','1','-subj','/CN=offline-test-manager-ca','-addext','basicConstraints=critical,CA:TRUE', '-addext','keyUsage=critical,keyCertSign,cRLSign') leaf,csr=prepare(root/'leaf','127.0.0.1') ext=root/'leaf.ext';ext.write_text('subjectAltName=IP:127.0.0.1\nextendedKeyUsage=serverAuth\nbasicConstraints=critical,CA:FALSE\n') command('x509','-req','-in',str(csr),'-CA',str(ca),'-CAkey',str(key),'-CAcreateserial', '-out',str(cert),'-days','1','-extfile',str(ext)) yield root,ca,leaf,csr,cert # Test-only trust/key material is removed, never copied into a repo fixture. shutil.rmtree(root) def test_tls_valid_manager_chain_and_csr_reuse(tls): _,ca,key,csr,cert=tls before=(key.read_bytes(),csr.read_bytes()) verify(key,cert,ca,'127.0.0.1') assert prepare(key.parent,'127.0.0.1')==(key,csr) assert before==(key.read_bytes(),csr.read_bytes()) def test_tls_rejects_san_wrong_ca_and_ca_as_leaf(tls,tmp_path): root,ca,key,csr,cert=tls with pytest.raises(TLSError):verify(key,cert,ca,'127.0.0.2') with pytest.raises(TLSError):prepare(key.parent,'127.0.0.2') with pytest.raises(TLSError):verify(key,cert,cert,'127.0.0.1') with pytest.raises(TLSError):verify(key,ca,ca,'127.0.0.1') wrong_key,_=prepare(tmp_path/'other','127.0.0.1') with pytest.raises(TLSError,match='TLS_KEY_MISMATCH'):verify(wrong_key,cert,ca,'127.0.0.1') def test_tls_rejects_world_readable_key_and_symlinks(tls,tmp_path): _,ca,key,_,cert=tls copy=tmp_path/'private.key';copy.write_bytes(key.read_bytes());copy.chmod(0o644) with pytest.raises(TLSError):verify(copy,cert,ca,'127.0.0.1') link=tmp_path/'link.key';link.symlink_to(key) with pytest.raises(TLSError):verify(link,cert,ca,'127.0.0.1') def test_config_rejects_http_and_loads_only_private_files(tls,tmp_path): import json _,ca,_,_,_=tls env={'AI_CONTROL_SERVER_ID':'AI-test','AI_CONTROL_MANAGER_CA_FILE':str(ca), 'AI_CONTROL_MANAGER_BASE_URL':'https://manager.example.invalid', 'AI_CONTROL_LOGIN_BODY_FILE':str(tmp_path/'login.json'), 'AI_CONTROL_PASSWORD_FILE':str(tmp_path/'password'), 'CONTEXTRA_CONTROL_TOKEN_FILE':str(tmp_path/'control'), 'CONTEXTRA_RELAY_TOKEN_FILE':str(tmp_path/'relay')} (tmp_path/'password').write_text('non-deployable fixture password') (tmp_path/'control').write_text(CONTROL);(tmp_path/'relay').write_text(RELAY) login={'loginId':'test','controlPort':9443,'advertisedIp':'127.0.0.1', 'redis':dict(host='redis',port=6379,db=0,username=None,password=None,tls=False), 'bboxStreamKey':'test','connectionConfigVersion':1,'aiControlCa':ca.read_text()} (tmp_path/'login.json').write_text(json.dumps(login)) cfg=Config.from_env(env);assert cfg.server_id=='AI-test' and CONTROL not in repr(cfg) env['AI_CONTROL_MANAGER_BASE_URL']='http://manager.example.invalid' with pytest.raises(ValueError):Config.from_env(env) @contextmanager def fake_upstream(): calls=[] class Handler(BaseHTTPRequestHandler): def log_message(self,*args):pass def send(self,status,body=b'',headers=()): self.send_response(status) for k,v in headers:self.send_header(k,v) self.send_header('Content-Length',str(len(body)));self.end_headers() if self.command!='HEAD':self.wfile.write(body) def do_GET(self): calls.append((self.command,self.path,self.headers.get('Authorization'))) if self.path=='/internal/media-auth': return self.send(204 if self.headers.get('Authorization')=='Bearer '+MANAGER else 401) if self.path.endswith('/clip.mp4'): if self.headers.get('Range')=='bytes=0-3':return self.send(206,b'abcd',[('Content-Range','bytes 0-3/8')]) return self.send(200,b'abcdefgh') if self.path.endswith('/thumbnail.jpg'): return self.send(200,b'thumbnail') return self.send(404) do_HEAD=do_GET def do_POST(self): calls.append((self.command,self.path,self.headers.get('Authorization'))) self.rfile.read(int(self.headers.get('Content-Length','0'))) self.send(200,b'{"result":"SUCCESS","cameraId":2}') server=ThreadingHTTPServer(('127.0.0.1',0),Handler) thread=threading.Thread(target=server.serve_forever,daemon=True);thread.start() try:yield server.server_port,calls finally:server.shutdown();server.server_close();thread.join(2) def test_real_nginx_tls_routing_and_protected_media(tls,tmp_path): nginx=shutil.which('nginx') if not nginx:pytest.skip('Nginx not installed') root,ca,key,csr,cert=tls with socket.socket() as s:s.bind(('127.0.0.1',0));port=s.getsockname()[1] with fake_upstream() as (upstream,calls): config=(ROOT/'docker/secure-gateway.conf.template').read_text() config=config.replace('include /tmp/media-access.conf;', 'auth_request /_bridge_media_auth;') config=config.replace('${AI_CONTROL_HOST_PORT}',str(port)).replace('listen '+str(port)+' ssl','listen 127.0.0.1:'+str(port)+' ssl') config=config.replace('/run/ai-tls/control-tls.crt',str(cert)).replace('/run/ai-tls/control-tls.key',str(key)) config=config.replace('http://ai-control-sidecar:8080',f'http://127.0.0.1:{upstream}').replace('http://analytics:5001',f'http://127.0.0.1:{upstream}') config=config.replace('/tmp/nginx.pid',str(tmp_path/'nginx.pid')).replace('/tmp/client_body',str(tmp_path/'client_body')).replace('/tmp/proxy_temp',str(tmp_path/'proxy_temp')) path=tmp_path/'nginx.conf';path.write_text(config) subprocess.run([nginx,'-t','-c',str(path)],check=True,capture_output=True,timeout=5) process=subprocess.Popen([nginx,'-c',str(path),'-g','daemon off;'],stdout=subprocess.DEVNULL,stderr=subprocess.DEVNULL) context=ssl.create_default_context(cafile=str(ca)) opener=urllib.request.build_opener(urllib.request.ProxyHandler({}),urllib.request.HTTPSHandler(context=context)) def request(path,method='GET',authorized=False,extra=None): headers={'Authorization':'Bearer '+MANAGER} if authorized else {} headers.update(extra or {}) req=urllib.request.Request(f'https://127.0.0.1:{port}'+path,method=method,headers=headers,data=b'{}' if method=='POST' else None) try: with opener.open(req,timeout=2) as r:return r.status,r.read() except urllib.error.HTTPError as e:return e.code,e.read() try: last_error=None for _ in range(50): try: if request('/nope')[0]==404:break except (OSError,urllib.error.URLError) as exc: last_error=exc time.sleep(.02) else:pytest.fail(f'Gateway did not become available: {last_error!r}; exit={process.poll()}') for path in ('/internal/events','/_bridge_media_auth','/api/v1/streams','/docs','/artifacts/.staging/job.json'): assert request(path)[0]==404 assert request('/api/v1/cameras/start','POST',True)[0]==200 for scenario in ('INTRUSION','CROWD'): media=f'/artifacts/cameras/2/2026-09-17/AI_120000.000+0900_{scenario}_test-event/clip.mp4' assert request(media)[0]==401 assert request(media,authorized=True)==(200,b'abcdefgh') assert request(media,authorized=True,extra={'Range':'bytes=0-3'})==(206,b'abcd') assert request(media,'HEAD',True)[0]==200 thumbnail=media.replace('clip.mp4','thumbnail.jpg') assert request(thumbnail)[0]==401 assert request(thumbnail,authorized=True)==(200,b'thumbnail') assert request(media.replace('clip.mp4','event.json'),authorized=True)[0]==404 # The Manager bearer is only used by the auth subrequest, never the file server. assert all(token is None for method,path,token in calls if path.endswith(('/clip.mp4','/thumbnail.jpg'))) # Wrong/untrusted chain must fail the TLS connection. with pytest.raises(urllib.error.URLError): urllib.request.build_opener(urllib.request.ProxyHandler({})).open(f'https://127.0.0.1:{port}/nope',timeout=2) finally: process.terminate() try:process.wait(timeout=3) except subprocess.TimeoutExpired:process.kill();process.wait(timeout=2) def test_compose_structure_and_entrypoint_syntax(): class Loader(yaml.SafeLoader):pass Loader.add_constructor('!override',lambda loader,node:loader.construct_sequence(node)) doc=yaml.load((ROOT/'deploy/secure-bridge.compose.yml').read_text(),Loader=Loader) services=doc['services'] assert set(services)=={'analytics','ai-control-sidecar','ai-gateway'} assert not services['ai-control-sidecar'].get('ports') assert not services['ai-control-sidecar'].get('depends_on') assert services['analytics']['environment']['CONTEXTRA_EVENT_ARTIFACT_ENABLED'].endswith(':-0}') assert services['analytics']['ports'][0].startswith('127.0.0.1:') assert len(services['ai-control-sidecar']['tmpfs'])==1 for key in ('ai-control-sidecar','ai-gateway'): assert services[key]['read_only'] and services[key]['cap_drop']==['ALL'] assert not services[key].get('runtime') and not services[key].get('network_mode') subprocess.run(['bash','-n',str(ROOT/'docker/entrypoint.sh')],check=True,timeout=5) script=(ROOT/'docker/entrypoint.sh').read_text() assert '0) APP_MODULE=app.main' in script and '1) APP_MODULE=app.secure_runtime' in script assert 'python3 -m "$APP_MODULE"' in script gateway=services['ai-gateway'] assert gateway['environment']['AI_ARTIFACT_ACCESS_POLICY'].endswith(':-lan_public}') assert gateway['entrypoint']==['/bin/sh','/etc/bridge/entrypoint.sh'] subprocess.run(['sh','-n',str(ROOT/'docker/secure-gateway-entrypoint.sh')],check=True,timeout=5) # This is structural YAML validation, NOT docker compose config execution. def test_media_policy_renderer(tmp_path): import os source=(ROOT/'docker/secure-gateway-entrypoint.sh').read_text() # Run the production policy selection with isolated paths and fake render/check binaries. script=tmp_path/'entry.sh' script.write_text(source.replace('/tmp/',str(tmp_path)+'/').replace('/etc/bridge/nginx.conf.template',str(ROOT/'docker/secure-gateway.conf.template'))) for name, content in {'envsubst':'#!/bin/sh\ncat\n','nginx':'#!/bin/sh\nexit 0\n'}.items(): executable=tmp_path/name;executable.write_text(content);executable.chmod(0o700) env=dict(os.environ,PATH=str(tmp_path)+':'+os.environ['PATH'],AI_CONTROL_HOST_PORT='38443') env.pop('AI_ARTIFACT_ACCESS_POLICY',None) subprocess.run(['sh',str(script),'--check'],env=env,check=True,timeout=5) assert (tmp_path/'media-access.conf').read_text()=='allow 192.168.50.0/24;\ndeny all;\nauth_request off;\n' env['AI_ARTIFACT_LAN_CIDR']='192.168.30.0/24' subprocess.run(['sh',str(script),'--check'],env=env,check=True,timeout=5) assert (tmp_path/'media-access.conf').read_text().startswith('allow 192.168.30.0/24;') for invalid in ('0.0.0.0/0','172.15.0.0/16','192.168.30.0/8','192.168.30.1/24', '192.168.30.0/24; allow all','192.168.30.0/24\n10.0.0.0/8'): env['AI_ARTIFACT_LAN_CIDR']=invalid assert subprocess.run(['sh',str(script),'--check'],env=env,capture_output=True,timeout=5).returncode!=0 env['AI_ARTIFACT_LAN_CIDR']='192.168.30.0/24' env['AI_ARTIFACT_ACCESS_POLICY']='authenticated' subprocess.run(['sh',str(script),'--check'],env=env,check=True,timeout=5) assert (tmp_path/'media-access.conf').read_text()=='auth_request /_bridge_media_auth;\n' env['AI_ARTIFACT_ACCESS_POLICY']='allow_everyone' assert subprocess.run(['sh',str(script),'--check'],env=env,capture_output=True,timeout=5).returncode!=0 template=(ROOT/'docker/secure-gateway.conf.template').read_text() assert 'real_ip_header' not in template and 'set_real_ip_from' not in template def test_outbound_transport_real_tls(tls): import asyncio from control_sidecar.transport import Transport from control_sidecar.protocol import BoundaryError _,ca,key,_,cert=tls calls=[] class Handler(BaseHTTPRequestHandler): def log_message(self,*a):pass def do_POST(self): calls.append((self.path,self.headers.get('Authorization'),self.rfile.read(int(self.headers['Content-Length'])))) body=b'{"result":true,"data":{"accepted":1,"duplicated":0}}' self.send_response(200);self.send_header('Content-Length',str(len(body)));self.end_headers();self.wfile.write(body) server=ThreadingHTTPServer(('127.0.0.1',0),Handler) context=ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER);context.load_cert_chain(str(cert),str(key)) server.socket=context.wrap_socket(server.socket,server_side=True) thread=threading.Thread(target=server.serve_forever,daemon=True);thread.start() async def check(): tx=Transport(str(ca)) try: code,raw=await tx.request('POST',f'https://127.0.0.1:{server.server_port}/api/v1/events',body=b'{"fixture":true}',token=MANAGER) assert code==200 and b'accepted' in raw finally:await tx.close() try:asyncio.run(check()) finally:server.shutdown();server.server_close();thread.join(2) assert calls==[('/api/v1/events','Bearer '+MANAGER,b'{"fixture":true}')]