"""44项安全实验：仅离线或127.0.0.1，不启动Docker/Nginx、不读取生产凭证。

Python 3.12；C024需要httpx，C025/026需要cryptography。
运行：python run_lab.py；也可 python run_lab.py C027。
"""
import collections
import copy
import datetime as dt
import hashlib
import io
import ipaddress
import json
import math
import os
from pathlib import Path
import random
import re
import socket
import ssl
import sys
import tempfile
import threading
from types import MappingProxyType
from email.utils import parsedate_to_datetime
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

HERE = Path(__file__).resolve().parent

def demand(actual, expected):
    assert actual == expected, {'actual': actual, 'expected': expected}
    return actual

def execute(n, d):
    e=d.get('expected',{})
    if n == 1:
        def audit(trace):
            indexed={}
            for node in trace:
                if node['id'] in indexed:raise ValueError('duplicate')
                indexed[node['id']]=node
            sent=[];success=set()
            for node in sorted(trace,key=lambda item:item['seq']):
                if node['kind']=='business':
                    if node['parent'] is not None:raise ValueError('invalid-root')
                    continue
                if node['parent'] not in indexed:raise ValueError('missing-parent')
                parent=indexed[node['parent']]
                if parent['business']!=node['business']:raise ValueError('cross-business')
                expected_parent={'sdk':'business','upstream':'sdk'}[node['kind']]
                if parent['kind']!=expected_parent or parent['seq']>=node['seq']:raise ValueError('invalid-parent')
                if node['kind']=='upstream':
                    if d['serial'] and node['business'] in success:raise ValueError('after-success')
                    sent.append(node['id'])
                    if node.get('outcome')=='success':success.add(node['business'])
            return sent
        dispatches=audit(d['trace']);bad=[]
        for kind in d['negative_cases']:
            trace=copy.deepcopy(d['trace'])
            if kind=='duplicate':trace[-1]['id']=trace[-2]['id']
            elif kind=='missing-parent':trace[-1]['parent']='absent'
            elif kind=='cross-business':trace[-1]['business']='another-business'
            elif kind=='after-success':trace.append({'business':'b1','id':'u3','parent':'s1','kind':'upstream','seq':4,'outcome':'failed'})
            try:audit(trace);raise AssertionError('反例未拒绝')
            except ValueError as ex:bad.append(str(ex))
        return demand({'dispatches':len(dispatches),'rejected':bad},e)|{'dispatch_ids':dispatches,'attempt_tree':d['trace'],'field_contract':'自定义；非OpenTelemetry格式认证'}
    if n == 2:
        now=dt.datetime.fromisoformat(d['now']);waits=[];decisions=[]
        for fields in d['cases']:
            values=[value for name,value in fields if name.lower()=='retry-after']
            if len(values)!=1:waits.append(None);decisions.append('ambiguous-field-count');continue
            raw=values[0]
            if re.fullmatch(r'[0-9]+',raw):wait=int(raw)
            elif re.fullmatch(r'[A-Z][a-z]{2}, [0-9]{2} [A-Z][a-z]{2} [0-9]{4} [0-9]{2}:[0-9]{2}:[0-9]{2} GMT',raw):
                when=parsedate_to_datetime(raw);wait=max(0,(when-now).total_seconds())
            else:wait=None
            waits.append(wait);decisions.append('single' if wait is not None else 'invalid-single')
        return demand({'waits':waits,'decisions':decisions},e)|{'raw_field_arrays':d['cases'],'policy':'重复字段保守隔离；日期单值只覆盖本夹具IMF-fixdate'}
    if n == 3:
        result=[sum(c.values())<=d['budget_ms'] for c in d['cases']]
        demand(result,e);return {'admit':result,'sent':sum(result)}
    if n == 4:
        rng=random.Random(d['seed']); cap=d['cap_ms'];count=d['count']
        full=[rng.uniform(0,cap) for _ in range(count)]
        rng=random.Random(d['seed']);equal=[cap/2+rng.uniform(0,cap/2) for _ in range(count)]
        schedules={'none':[cap]*count,'full':full,'equal':equal}; result={}
        for name,seq in schedules.items():
            assert len(seq)==count and all(0<=x<=cap for x in seq)
            result[name]={'peak_buckets':{str(w):max(collections.Counter(int(x//w) for x in seq).values()) for w in d['bucket_ms']},'p95_wait_ms':sorted(seq)[math.ceil(count*.95)-1]}
        assert result['none']['peak_buckets']['100']==100
        assert result['full']['peak_buckets']['100']<100
        return {'comparison':result,'schedule_ms':schedules}
    if n == 5:
        buckets=collections.defaultdict(list)
        for raw in d['authoritative_ids']:
            candidate=str(int(raw)) if raw.isdecimal() else raw.lower()
            buckets[candidate].append(raw)
        collisions={key:values for key,values in buckets.items() if len(values)>1}
        def safe_key(raw):
            if raw not in d['authoritative_ids']:raise ValueError('unknown-authoritative-id')
            return ('tenant',raw)
        safe={safe_key(raw) for raw in d['authoritative_ids']}
        try:safe_key(d['unknown']);unknown=False
        except ValueError:unknown=True
        return demand({'collisions':collisions,'safe_bucket_count':len(safe),'unknown_rejected':unknown},e)|{'candidate_mapping':dict(buckets)}
    if n == 6:
        fixed=collections.Counter();rolling=[];f=[];r=[]
        for i,t in enumerate(d['arrivals_ms']):
            bucket=t//d['width_ms']
            if fixed[bucket]<d['limit']:fixed[bucket]+=1;f.append(i)
            rolling=[x for x in rolling if x>t-d['width_ms']]
            if len(rolling)<d['limit']:rolling.append(t);r.append(i)
        return demand({'fixed':f,'rolling':r},e)
    if n == 7:
        reserve=dict(d['reserve']);available=d['capacity']-sum(reserve.values());settled=0;history=[]
        for action,key,usage in d['events']:
            if key in reserve and action in ('settle','cancel'):
                amount=reserve.pop(key)
                assert usage<=amount
                available+=amount-usage;settled+=usage
            assert available+sum(reserve.values())+settled==d['capacity']
            history.append([available,sum(reserve.values()),settled])
        return demand({'available':available,'reserved':sum(reserve.values()),'settled':settled},e)|{'conservation':history}
    if n == 8:
        def waits(jobs):
            now=0;result={}
            for job in jobs:result[job['id']]=[now,now+job['ms']];now+=job['ms']
            return result
        grouped=collections.defaultdict(collections.deque)
        for job in d['jobs']:grouped[job['tenant']].append(job)
        rr=[]
        while any(grouped.values()):
            for q in grouped.values():
                if q:rr.append(q.popleft())
        fifo=waits(d['jobs']);fair=waits(rr)
        return demand({'fifo_b1_wait':fifo['b1'][0],'round_robin_b1_wait':fair['b1'][0],'long_finish':fair['long'][1]},e)|{'fifo':fifo,'round_robin':fair}
    if n == 9:
        active=('v1',d['versions']['v1']);limit_version=active[0];mixed_limit=active[1]['limit']
        if d['switch_between_fields']:active=('v2',d['versions']['v2'])
        window_version=active[0];mixed_window=active[1]['window_s'];rejected=limit_version!=window_version
        active=('v1',d['versions']['v1']);version=active[0];snapshot=MappingProxyType(dict(active[1]))
        active=('v2',d['versions']['v2'])
        def admit(config):return sum(t>d['now_s']-config['window_s'] for t in d['prior_s'])<config['limit']
        result={'mixed_rejected':rejected,'snapshot_version':version,'snapshot_admit':admit(snapshot),'new_request_version':active[0],'new_request_admit':admit(active[1])}
        try:snapshot['limit']=999;raise AssertionError('快照应不可变')
        except TypeError:pass
        return demand(result,e)|{'mixed_read':{'limit_version':limit_version,'limit':mixed_limit,'window_version':window_version,'window_s':mixed_window},'snapshot':dict(snapshot),'scope':'交错事件参考，不是实际线程原子性验证'}
    if n == 10:
        expiry=dt.datetime.fromisoformat(d['expiry']);results=[]
        for raw in d['now']:
            t=dt.datetime.fromisoformat(raw)
            results.append('invalid' if t.tzinfo is None else t<expiry)
        demand(results,e);return {'allowed':results}
    if n == 11:
        dispatches=[];trace=[]
        def fake_upstream(target):
            dispatches.append(target)
            return {'synthetic_target':target}
        def replay(mapping,phase):
            decisions=[]
            for requested in d['requests']:
                target=mapping[requested]
                allowed=requested in d['allow_alias'] and target in d['allow_targets']
                calls_before=len(dispatches)
                response=fake_upstream(target) if allowed else None
                delta=len(dispatches)-calls_before
                assert delta==(1 if allowed else 0)
                trace.append({'phase':phase,'requested':requested,'resolved':target,'allowed':allowed,'dispatch_count':delta,'response':response})
                decisions.append(allowed)
            return decisions
        result={'before':replay(d['aliases'],'before'),'after':replay(d['changed_aliases'],'after')}
        demand(result,e)
        denied=[x['dispatch_count'] for x in trace if not x['allowed']]
        assert len(denied)==3 and all(x==0 for x in denied)
        return result|{'denied_upstream_calls':sum(denied),'allowed_upstream_calls':len(dispatches),'dispatch_trace':trace}
    if n == 12:
        expiry={k:v+d['ttl_ms'] for k,v in d['loaded_ms'].items()};rev=d['revoked_ms']
        result={'ttl':max(expiry.values())-rev,'event_r1':d['invalidate_ms']['r1']-rev,'lost_r2':expiry['r2']-rev}
        demand(result,d['expected_stale_ms'])
        return result|{'admit_without_event':{k:[t<exp for t in d['queries_ms']] for k,exp in expiry.items()}}
    if n == 13:
        result=[d['request'] if uri=='' else uri+d['request'][len(d['location']):] for uri in d['upstream_uri']]
        demand(result,e);assert not d['negative'].startswith(d['location'])
        assert 'proxy_pass http://127.0.0.1:8999/;' in (HERE/'c013.conf').read_text()
        return {'reference_paths':result,'nginx_executed':False}
    if n == 14:
        assert d['status']==200 and d['content_encoding'] is None and d['encoding']=='utf-8'
        before=d['before'].encode(d['encoding']);after=d['after'].encode(d['encoding']);old=len(before);characters=len(d['after'])
        validate=lambda header:header==len(after)
        return demand({'before_bytes':len(before),'after_characters':characters,'after_bytes':len(after),'old_length_valid':validate(old),'character_length_valid':validate(characters),'recomputed_valid':validate(len(after))},e)|{'metadata_action':'删除旧Content-Length，将最终固定字节交给HTTP库重算；未测试协议栈实际发送'}
    if n == 15:
        class Tailer:
            def __init__(self):self.pending=bytearray();self.records=[]
            def feed(self,chunk):
                self.pending.extend(chunk)
                while b'\n' in self.pending:
                    line,_,remaining=self.pending.partition(b'\n')
                    value=json.loads(line.decode('utf-8'))
                    if not isinstance(value,dict):raise ValueError('本采集合同仅接收对象')
                    self.records.append(value);self.pending=bytearray(remaining)
        first=(json.dumps(d['complete_record'],ensure_ascii=False,separators=(',',':'))+'\n').encode()+d['pending_text'].encode();suffix=d['append_text'].encode();tail=Tailer();tail.feed(first)
        before=len(tail.records);pending=bytes(tail.pending);assert pending==d['pending_text'].encode()
        tail.feed(b'');assert len(tail.records)==before and bytes(tail.pending)==pending
        tail.feed(suffix);after=len(tail.records);tail.feed(b'');assert not tail.pending
        combined=first+suffix
        for chunks in [[combined],[bytes([byte]) for byte in combined],[combined[:7],combined[7:22],combined[22:]]]:
            other=Tailer()
            for chunk in chunks:other.feed(chunk)
            assert other.records==tail.records and not other.pending
        no_lf=Tailer();no_lf.feed(b'{"id":"complete"}');waits=len(no_lf.records)==0 and bool(no_lf.pending);no_lf.feed(b'\n');assert len(no_lf.records)==1
        return demand({'committed_before':before,'committed_after':after,'ids':[row['id'] for row in tail.records],'complete_without_lf_waits':waits},e)|{'pending_hex_before_append':pending.hex(),'pending_bytes':len(pending),'chunk_layouts_checked':3,'contract':'持续追加tailer以LF提交；JSON Lines标准允许完整末行无LF'}
    if n == 16:
        def simulate(cap):
            produced=consumed=queued=high=0;t=0
            while consumed<d['produced_chunks']:
                if produced<d['produced_chunks'] and (cap is None or queued+d['chunk_bytes']<=cap):
                    produced+=1;queued+=d['chunk_bytes'];high=max(high,queued)
                if t not in d['consumer_pause_ticks'] and queued:
                    queued-=d['chunk_bytes'];consumed+=1
                assert queued>=0 and (cap is None or queued<=cap)
                t+=1;assert t<100
            return high,consumed*d['chunk_bytes']
        bounded,delivered=simulate(d['buffer_cap']);unbounded,_=simulate(None)
        return demand({'bounded_high_water':bounded,'unbounded_high_water':unbounded,'delivered':delivered},e)
    if n == 17:
        def replay(propagate):
            state={'closed':False,'produced':0};pending=collections.deque();sent=[]
            def upstream():
                try:
                    for i in range(d['chunks']):state['produced']+=1;yield i
                finally:state['closed']=True
            stream=upstream()
            for _ in range(d['disconnect_after']):
                pending.append(next(stream));sent.append(pending.popleft())
            pending.append({'late_buffered':'synthetic-before-cleanup'})
            queue_before=len(pending)
            if propagate:stream.close();pending.clear()
            else:list(stream)
            return state|{'queue_before_cleanup':queue_before,'queue_after_cleanup':len(pending),'sent':sent}
        ignored=replay(False);state=replay(True)
        result={'ignore_produced':ignored['produced'],'propagate_produced':state['produced'],'closed':state['closed'],'queued':state['queue_after_cleanup']}
        return demand(result,e)|{'queue_before_cleanup':state['queue_before_cleanup'],'ignored_queue_after':ignored['queue_after_cleanup'],'delivered':state['sent']}
    if n == 18:
        sizes=[len(x.encode('utf-8')) for x in d['payloads']];allow=[s<=d['limit_bytes'] for s in sizes]
        dispatch=[];trace=[]
        def fake_upstream(body):dispatch.append(body);return len(body)
        for payload,admitted in zip(d['payloads'],allow):
            before=len(dispatch)
            if admitted:fake_upstream(payload.encode('utf-8'))
            delta=len(dispatch)-before
            trace.append({'accepted':admitted,'upstream_calls':delta})
            assert delta==(1 if admitted else 0)
        return demand({'lengths':sizes,'accepted':allow,'upstream_calls':len(dispatch)},e)|{'dispatch_trace':trace}
    if n == 19:
        networks=[ipaddress.ip_network(x) for x in d['trusted']]
        def trusted(addr):return any(addr in net for net in networks if addr.version==net.version)
        def resolve(v):
            peer=ipaddress.ip_address(v['peer'])
            if not trusted(peer):return str(peer)
            try:chain=[ipaddress.ip_address(x.strip()) for x in v['xff'].split(',')]
            except ValueError:return str(peer)
            current=peer
            for item in reversed(chain):
                if not trusted(current):break
                current=item
            return str(current)
        result=[resolve(x) for x in d['vectors']];demand(result,e);return {'client_ip':result}
    if n == 20:
        tokens={part.strip().lower() for key,value in d['headers'] if key.lower()=='connection' for part in value.split(',')}
        hop=tokens|{'connection','keep-alive','proxy-authenticate','proxy-authorization','te','trailer','transfer-encoding','upgrade'}
        result=[x for x in d['headers'] if x[0].lower() not in hop]
        demand(result,e);return {'forwarded':result,'removed_names':sorted(hop)}
    if n == 21:
        def substitute(raw):
            required=re.fullmatch(r'\$\{([A-Za-z_][A-Za-z_0-9]*):\?[^}]+\}',raw)
            if required:return d['variables'].get(required[1]) or 'required-error'
            sentinel='__LITERAL_DOLLAR__';raw=raw.replace('$$',sentinel)
            raw=re.sub(r'\$([A-Za-z_][A-Za-z_0-9]*)',lambda m:d['variables'].get(m[1],''),raw)
            return raw.replace(sentinel,'$')
        result=[substitute(x) for x in d['expressions']];demand(result,e)
        assert 'demo$$DEMO' in (HERE/'c021.compose.yaml').read_text()
        return {'reference_rendered':result,'compose_executed':False}
    if n == 22:
        v=d['events'];ready=lambda t:v['query_ready_ms']<=t<v['failure_ms']
        config=(HERE/'c022.compose.yaml').read_text('utf-8')
        required=['depends_on:','condition: service_healthy','healthcheck:','local_query_probe.py','example.invalid/not-pulled:fixture']
        assert all(part in config for part in required)
        return demand({'early_ready':ready(d['app_start_ms']['started_condition']),'healthy_ready':ready(d['app_start_ms']['healthy_condition']),'later_ready':ready(v['failure_ms'])},e)|{'static_tokens_checked':required,'compose_executed':False,'yaml_semantics_validated':False}
    if n == 23:
        def classify(item):
            if 'host_ip' not in item:return 'unspecified-all'
            address=ipaddress.ip_address(item['host_ip']);family='ipv'+str(address.version)
            return family+('-loopback' if address.is_loopback else ('-wildcard' if address.is_unspecified else '-specific-host'))
        result=[classify(item) for item in d['ports']];demand(result,d['expected_scope'])
        assert len(set(result[3:]))==3
        config=(HERE/'c023.compose.yaml').read_text('utf-8')
        checked=['127.0.0.1','0.0.0.0','192.0.2.8','"::"']
        assert all('host_ip: '+host in config for host in checked)
        return {'configuration_scopes':result,'address_properties':[{'host':x.get('host_ip'),'is_loopback':ipaddress.ip_address(x['host_ip']).is_loopback if 'host_ip' in x else None,'is_unspecified':ipaddress.ip_address(x['host_ip']).is_unspecified if 'host_ip' in x else None} for x in d['ports']],'static_host_tokens_checked':checked,'docker_listener_executed':False,'yaml_semantics_validated':False}
    if n == 24:return proxy_environment_lab(d)
    if n in (25,26):return tls_lab(n,d)
    if n == 27:
        def compatible(bounds,window):
            if bounds!=d['boundaries_ms'] or window!=d['window_s']:raise ValueError('incompatible-histogram')
        compatible(d['boundaries_ms'],d['window_s'])
        refused=0
        for bounds,window in [(d['incompatible_boundaries_ms'],d['window_s']),(d['boundaries_ms'],d['incompatible_window_s'])]:
            try:compatible(bounds,window)
            except ValueError:refused+=1
        assert refused==2
        def p95(values):return sorted(values)[math.ceil(len(values)*.95)-1]
        replicas=list(d['latencies_ms'].values());ps=[p95(v) for v in replicas];values=sum(replicas,[])
        buckets={str(b):sum(x<=b for x in values) for b in d['boundaries_ms']};q=.95*len(values);last_b=last_c=0
        for bound in d['boundaries_ms']:
            count=buckets[str(bound)]
            if count>=q:
                estimate=last_b+(q-last_c)/(count-last_c)*(bound-last_b);break
            last_b,last_c=bound,count
        return demand({'replica_p95':ps,'mean_p95':sum(ps)/len(ps),'global_p95':p95(values),'histogram_estimate':estimate},e)|{'cumulative_buckets':buckets|{'+Inf':len(values)},'incompatible_inputs_refused':refused}
    if n == 28:
        jsonl=''.join(json.dumps(x,ensure_ascii=False)+'\n' for x in d['events'])
        raw=''.join(x['id']+' '+x['marker']+'\n' for x in d['events'])
        demand([json.loads(x) for x in jsonl.splitlines()],d['events'])
        return demand({'jsonl_lines':len(jsonl.splitlines()),'raw_lines':len(raw.splitlines())},e)|{'jsonl':jsonl}
    if n == 29:
        order=['image','env_file','environment','cli'];result=[d[next(k for k in reversed(order) if k in case)] for case in d['cases']]
        demand(result,e)
        config=(HERE/'c029.compose.yaml').read_text('utf-8');env=(HERE/'c029.env').read_text('utf-8')
        assert 'env_file: [c029.env]' in config and 'MODE: '+d['environment'] in config
        assert env.strip()=='MODE='+d['env_file']
        return {'reference_final_values':result,'static_config_and_env_checked':True,'compose_executed':False,'yaml_semantics_validated':False}
    if n == 30:
        def resolve(rows):
            mapping={}
            for key,value in rows:
                if key in mapping:raise ValueError('duplicate')
                mapping[key]=value
            resolved={}
            for start in mapping:
                seen=set();key=start
                while key not in d['targets']:
                    if key in seen:raise ValueError('cycle')
                    seen.add(key)
                    if key not in mapping:raise ValueError('dangling')
                    key=mapping[key]
                resolved[start]=key
            return resolved
        resolved=resolve(d['valid']);bad=[]
        for kind in ['duplicate','dangling','cycle']:
            try:resolve(d[kind]);raise AssertionError('应当拒绝')
            except ValueError as ex:bad.append(str(ex))
        return demand({'resolved':resolved,'invalid':bad},e)
    if n == 31:
        def schedule(weights,count):
            current=dict.fromkeys(weights,0);sequence=[]
            for _ in range(count):
                for key,value in weights.items():current[key]+=value
                selected=max(weights,key=lambda k:current[k]);current[selected]-=sum(weights.values());sequence.append(selected)
            return sequence
        eight=schedule(d['weights'],8);long=schedule(d['weights'],400);disabled=schedule({'b':3},20)
        return demand({'eight':dict(collections.Counter(eight)),'four_hundred':dict(collections.Counter(long)),'disabled_a':disabled.count(d['disabled'])},e)|{'first_eight':eight}
    if n == 32:
        snapshots=[0 for _ in range(d['arrivals'])]
        unlocked=[i for i,snapshot in enumerate(snapshots) if snapshot<d['probe_cap']]
        permits=0;admitted=0
        for _ in range(d['arrivals']):
            if permits<d['probe_cap']:permits+=1;admitted+=1
        states=[];outcome_trace=[]
        for outcome in d['outcomes']:
            in_use=0;state='HALF_OPEN'
            in_use+=1;before=in_use
            try:
                if outcome=='success':state='CLOSED'
                elif outcome=='failure':state='OPEN'
                elif outcome=='cancel':raise InterruptedError('合成取消')
                else:raise ValueError(outcome)
            except InterruptedError:state='HALF_OPEN'
            finally:in_use-=1
            assert before==1 and in_use==0
            states.append(state);outcome_trace.append({'outcome':outcome,'permits_before':before,'permits_after':in_use,'state':state})
        after_total=sum(item['permits_after'] for item in outcome_trace)
        return demand({'unlocked':len(unlocked),'atomic':admitted,'final_states':states,'permits_after':after_total},e)|{'unlocked_read_snapshots':snapshots,'outcome_trace':outcome_trace}
    if n == 33:
        failed=next(x for x in d['channels'] if x['id']==d['failed']);result={}
        for label,key in [('entity','id'),('display','display'),('account','account')]:result[label]=[x['id'] for x in d['channels'] if x[key]==failed[key]]
        return demand(result,e)
    if n == 34:
        def replay(fall,rise):
            up=True;fails=successes=flips=0;down_at=None;recovery=None
            for i,ok in enumerate(d['probes']):
                successes=successes+1 if ok else 0;fails=0 if ok else fails+1
                if up and fails>=fall:up=False;flips+=1;down_at=i
                elif not up and successes>=rise:up=True;flips+=1;recovery=i
            return flips,down_at,recovery
        instant,_,instant_rec=replay(1,1);hys,down,recovery=replay(d['fall'],d['rise'])
        return demand({'instant_flips':instant,'hysteresis_flips':hys,'first_down_index':down,'recovery_extra_ms':(recovery-instant_rec)*d['interval_ms']},e)
    if n == 35:
        accepted={};used=0
        for job in d['jobs']:
            if len(accepted)<d['count_cap'] and used+job['bytes']<=d['byte_cap']:accepted[job['id']]=job['bytes'];used+=job['bytes']
            assert used<=d['byte_cap']
        count=len(accepted);used-=accepted.pop(d['cancel']);assert used==sum(accepted.values())
        used-=accepted.pop(d['cancel'],0);assert used==sum(accepted.values())
        return demand({'count_only_bytes':sum(x['bytes'] for x in d['jobs']),'accepted':count,'after_cancel_bytes':used},e)
    if n == 36:
        result={'completed':[],'rejected':[],'cancelled':[]}
        for job in d['jobs']:
            category='rejected' if job['start']>=d['drain_ms'] else ('completed' if job['finish']<=d['deadline_ms'] else 'cancelled')
            result[category].append(job['id'])
        return demand(result,e)
    if n == 37:
        detected=[bool(re.search(r'^exec python ',text,re.M)) for text in [d['shell'],d['exec']]]
        for name,key in [('c037-shell.sh','shell'),('c037-exec.sh','exec')]:demand((HERE/name).read_text(),d[key])
        return demand({'exec_detected':detected,'budget_fits':d['cleanup_budget_s']<d['stop_grace_s']},e)|{'docker_signal_executed':False}
    if n == 38:
        def parse(raw):
            if type(raw) is not str:return 'invalid-type'
            if raw=='true':return True
            if raw=='false':return False
            return 'invalid'
        return demand({'parsed':[parse(raw) for raw in d['values']],'string_truth':[bool(raw) for raw in d['values']]},e)|{'raw_values':d['values'],'grammar':'仅精确小写true/false；不裁空白、不强制字符串化；非产品默认'}
    if n == 39:
        unsafe={};safe={};uh=sh=cross=0
        for tenant,perm,config in d['requests']:
            payload_hash=hashlib.sha256(d['payload'].encode()).hexdigest()
            key=(tenant,perm,config,payload_hash)
            if payload_hash in unsafe:uh+=1
            else:unsafe[payload_hash]=tenant
            if key in safe:
                sh+=1;cross+=safe[key]!=tenant
            else:safe[key]=tenant
        return demand({'unsafe_hits':uh,'safe_hits':sh,'cross_tenant_safe_hits':cross},e)
    if n == 40:
        def replay(cache_errors,purge=False):
            saved=None;expiry=0;upstream=0;statuses=[]
            for index,t in enumerate(d['requests_ms']):
                if purge and index==1:saved=None
                if saved is not None and t<expiry:statuses.append(saved);continue
                status=d['upstream_statuses'][upstream];upstream+=1;statuses.append(status)
                if status==200 or cache_errors:saved=status;expiry=t+d['ttl_ms']
            return statuses,upstream
        cached,c1=replay(True);uncached,c2=replay(False);purged,c3=replay(True,True)
        return demand({'cache_error':cached,'no_error_cache':uncached,'purge_before_second':purged},e)|{'reference_upstream_calls':[c1,c2,c3],'nginx_executed':False}
    if n == 41:
        data=d['body'].encode();assert len(data)<=d['max_replay_bytes'];stream=io.BytesIO(data);first=stream.read();second=stream.read();stream.seek(0);third=stream.read();fourth=io.BytesIO(data).read()
        class OnePass(io.RawIOBase):
            def __init__(self,payload):self.content=io.BytesIO(payload);self.seek_attempts=0
            def readable(self):return True
            def seekable(self):return False
            def read(self,size=-1):return self.content.read(size)
            def seek(self,*args):self.seek_attempts+=1;raise io.UnsupportedOperation('one-pass fixture')
        source=OnePass(data);one_pass_first=source.read();assert one_pass_first==data
        try:source.seek(0);seek_error=False
        except io.UnsupportedOperation:seek_error=True
        calls=[]
        def retry_body(candidate):
            if not candidate.seekable():return 'deny'
            candidate.seek(0);calls.append(candidate.read());return 'sent'
        decision=retry_body(source)
        assert seek_error and not source.seekable() and source.seek_attempts==1 and not calls
        return demand({'same_object_second_length':len(second),'rewound_equal':first==third,'rebuilt_equal':first==fourth,'unseekable_retry':decision},e)|{'length':len(data),'sha256':hashlib.sha256(first).hexdigest(),'one_pass_seekable':source.seekable(),'seek_error_observed':seek_error,'retry_dispatches':len(calls)}
    if n == 42:
        rows=d['records'];assert all(x['synthetic'] and re.fullmatch('sha256:[0-9a-f]{64}',x['digest']) for x in rows)
        return demand({'same_tag':len({x['tag'] for x in rows})==1,'same_digest':len({x['digest'] for x in rows})==1,'platforms':len({x['platform'] for x in rows})},e)
    if n == 43:
        assert all(not x['write'] for x in d['paths'] if x['kind']=='root')
        assert 'read_only: true' in (HERE/'c043.compose.yaml').read_text()
        return demand({'writable':sum(x['write'] for x in d['paths']),'root_denied':sum(x['kind']=='root' and not x['write'] for x in d['paths'])},e)|{'docker_writes_executed':False}
    if n == 44:
        wrapper={'log':d['message'],'stream':d['stream'],'time':d['time']};encoded=(json.dumps(wrapper,ensure_ascii=False,separators=(',',':'))+'\n').encode();line_bytes=len(encoded)
        rate=line_bytes*d['lines_per_second']*d['containers'];rotation=d['max_size_bytes']*d['max_file']*d['containers']
        assert json.loads(encoded)==wrapper and line_bytes>len(d['message'].encode())
        actual=(HERE/'c044-synthetic.jsonl').read_bytes();assert actual==encoded and b'\r' not in actual and actual.endswith(b'\n')
        return {'synthetic_line_bytes':line_bytes,'actual_file_bytes':len(actual),'file_matches_calculation':actual==encoded,'growth_bytes_per_second':rate,'growth_bytes_per_hour':rate*3600,'nominal_rotation_bytes':rotation,'budget_with_headroom':math.ceil(rotation*d['headroom']),'unrotated_fill_seconds':rotation/rate,'docker_executed':False}
    raise ValueError(n)

def proxy_environment_lab(d):
    import httpx
    counts=collections.Counter()
    def start(label):
        class Handler(BaseHTTPRequestHandler):
            def do_GET(self):
                counts[label]+=1;data=label.encode();self.send_response(200);self.send_header('Content-Length',str(len(data)));self.end_headers();self.wfile.write(data)
            def log_message(self,*args):pass
        server=ThreadingHTTPServer(('127.0.0.1',0),Handler);thread=threading.Thread(target=server.serve_forever,daemon=True);thread.start();return server,thread
    target,tt=start('target');proxy,pt=start('proxy');keys=['HTTP_PROXY','HTTPS_PROXY','ALL_PROXY','NO_PROXY','http_proxy','https_proxy','all_proxy','no_proxy'];backup={k:os.environ.get(k) for k in keys}
    try:
        for key in keys:os.environ.pop(key,None)
        os.environ['HTTP_PROXY']=f'http://127.0.0.1:{proxy.server_port}'
        url=f'http://127.0.0.1:{target.server_port}'+d['url_path']
        with httpx.Client(timeout=2) as client:a=client.get(url).text
        os.environ['NO_PROXY']='127.0.0.1'
        with httpx.Client(timeout=2) as client:b=client.get(url).text
        os.environ.pop('NO_PROXY')
        with httpx.Client(trust_env=False,timeout=2) as client:c=client.get(url).text
        result={'environment_proxy':a,'no_proxy':b,'trust_env_false':c};demand(result,d['expected'])
        demand(dict(counts),{'proxy':1,'target':2});return result|{'hit_counts':dict(counts),'httpx_version':httpx.__version__,'scope':'127.0.0.1-only'}
    finally:
        for key,value in backup.items():
            if value is None:os.environ.pop(key,None)
            else:os.environ[key]=value
        for server,thread in [(target,tt),(proxy,pt)]:server.shutdown();server.server_close();thread.join(timeout=2)

def tls_lab(n,d):
    from cryptography import x509
    from cryptography.hazmat.primitives import hashes,serialization
    from cryptography.hazmat.primitives.asymmetric import rsa
    from cryptography.x509.oid import NameOID,ExtendedKeyUsageOID
    def issue(name,issuer=None,ca=False,dns=None):
        key=rsa.generate_private_key(public_exponent=65537,key_size=2048)
        subject=x509.Name([x509.NameAttribute(NameOID.COMMON_NAME,name)])
        issuer_name=issuer[0].subject if issuer else subject;signer=issuer[1] if issuer else key
        builder=x509.CertificateBuilder().subject_name(subject).issuer_name(issuer_name).public_key(key.public_key()).serial_number(x509.random_serial_number()).not_valid_before(dt.datetime(2020,1,1,tzinfo=dt.timezone.utc)).not_valid_after(dt.datetime(2040,1,1,tzinfo=dt.timezone.utc)).add_extension(x509.BasicConstraints(ca=ca,path_length=None),critical=True)
        if dns:builder=builder.add_extension(x509.SubjectAlternativeName([x509.DNSName(dns)]),critical=False).add_extension(x509.ExtendedKeyUsage([ExtendedKeyUsageOID.SERVER_AUTH]),critical=False)
        return builder.sign(signer,hashes.SHA256()),key
    root=issue('fixture-root',ca=True)
    intermediate=issue('fixture-intermediate',issuer=root,ca=True) if n==26 else None
    leaf=issue(d['dns_name'],issuer=intermediate or root,dns=d['dns_name'])
    pem=lambda cert:cert.public_bytes(serialization.Encoding.PEM)
    results=[]
    with tempfile.TemporaryDirectory(prefix='gateway-tls-fixture-') as temp:
        folder=Path(temp);ca=folder/'root.pem';ca.write_bytes(pem(root[0]));key=folder/'leaf.key';key.write_bytes(leaf[1].private_bytes(serialization.Encoding.PEM,serialization.PrivateFormat.PKCS8,serialization.NoEncryption()))
        cases=d['names'] if n==25 else d['chains']
        for case in cases:
            cert=folder/'server.pem';cert.write_bytes(pem(leaf[0])+(pem(intermediate[0]) if n==26 and case=='leaf_plus_intermediate' else b''))
            ctx=ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER);ctx.load_cert_chain(str(cert),str(key))
            listener=socket.socket();listener.bind(('127.0.0.1',0));listener.listen(1);listener.settimeout(3);port=listener.getsockname()[1]
            def serve():
                try:
                    connection,_=listener.accept()
                    with connection:
                        try:
                            with ctx.wrap_socket(connection,server_side=True) as secure:secure.sendall(b'ok')
                        except (ssl.SSLError,ConnectionResetError,BrokenPipeError):pass
                finally:listener.close()
            thread=threading.Thread(target=serve,daemon=True);thread.start()
            client=ssl.create_default_context(cafile=str(ca))
            try:
                with socket.create_connection(('127.0.0.1',port),timeout=2) as raw:
                    with client.wrap_socket(raw,server_hostname=case if n==25 else d['dns_name']) as secure:results.append(secure.recv(2)==b'ok')
            except ssl.SSLCertVerificationError:results.append(False)
            thread.join(timeout=3);assert not thread.is_alive()
    demand(results,d['expected'])
    return {'verified':results,'openssl':ssl.OPENSSL_VERSION,'scope':'loopback_TLS','keys':'临时生成，退出实验目录后已删除，未更改系统信任库'}

def run(ids=None):
    result=[]
    for n in range(1,45):
        article=f'C{n:03d}'
        if ids and article not in ids:continue
        data=json.loads((HERE/f'c{n:03d}.json').read_text('utf-8'))
        assert data['id']==article and data['synthetic'] is True
        level='loopback_tls' if n in (25,26) else ('loopback_http' if n==24 else ('static_configuration_reference' if n in (13,21,22,23,29,37,40,43) else 'offline_fixture'))
        try:details=execute(n,data['input']);result.append({'id':article,'method':level,'passed':True,'result':details})
        except Exception as ex:result.append({'id':article,'method':level,'passed':False,'error':repr(ex)})
    return {'passed':all(x['passed'] for x in result),'article_count':len(result),'checks':result,'python_version':sys.version.split()[0],'nginx_run':False,'docker_run':False,'external_network':False}

if __name__=='__main__':
    output=run(set(sys.argv[1:]) if len(sys.argv)>1 else None)
    print(json.dumps(output,ensure_ascii=False,indent=2));raise SystemExit(0 if output['passed'] else 1)