From 79c5af18532bc96395404d4f78b532da9e6a63be Mon Sep 17 00:00:00 2001 From: fy59 Date: Sat, 17 Oct 2026 07:50:03 +0200 Subject: [PATCH] feat(probe): implement gated cgroup task admission --- docs/KNOWN_LIMITATIONS.md | 11 +- docs/README.md | 8 +- docs/ROADMAP.md | 12 +- .../STEP20C_EGL_FX_TASK_LIMIT_PROTOCOL.md | 139 ++- docs/design/STEP20C_HEADLESS_EGL_FX_PROBE.md | 43 +- tooling/probes/step20c-egl-fx/README.md | 130 ++- tooling/probes/step20c-egl-fx/cgroup_gate.py | 724 +++++++++++++ tooling/probes/step20c-egl-fx/prepare.py | 102 +- tooling/probes/step20c-egl-fx/run.py | 106 +- .../step20c-egl-fx/test_cgroup_admission.py | 957 ++++++++++++++++++ tooling/probes/step20c-egl-fx/test_probe.py | 82 +- 11 files changed, 2199 insertions(+), 115 deletions(-) create mode 100644 tooling/probes/step20c-egl-fx/cgroup_gate.py create mode 100644 tooling/probes/step20c-egl-fx/test_cgroup_admission.py diff --git a/docs/KNOWN_LIMITATIONS.md b/docs/KNOWN_LIMITATIONS.md index 25dcd4e..b5c8f19 100644 --- a/docs/KNOWN_LIMITATIONS.md +++ b/docs/KNOWN_LIMITATIONS.md @@ -2,10 +2,13 @@ The [EGL/FX attempt](design/STEP20C_HEADLESS_EGL_FX_PROBE.md#observed-single-attempt--2026-10-17) failed at Bubblewrap namespace creation before the helper: EGL_NOT_RUN / -ALLOCATOR_NOT_ATTEMPTED; budget consumed. A probe-local cgroup task-limit -[protocol is DESIGN_ONLY](design/STEP20C_EGL_FX_TASK_LIMIT_PROTOCOL.md), -EGL_FX_RETRY_PREP_BLOCKED; new runtime NOT_RUN. No production contract or -STEP20C status changes, and no new native execution is authorized. +ALLOCATOR_NOT_ATTEMPTED; budget consumed. The probe-local cgroup task-limit +[protocol is IMPLEMENTED_OFFLINE](design/STEP20C_EGL_FX_TASK_LIMIT_PROTOCOL.md), +EXPERIMENTAL / OFFLINE_REVIEW_ACCEPTED, with inert validation only. Real manager +qualification, including parent-death cleanup, and new EGL runtime remain +NOT_RUN. Separate approvals are required for benign qualification and any later +EGL attempt. No production contract or STEP20C status changes, and no new native +execution is authorized. The [CORE-C V2 follow-up decision V4](design/STEP20C_COREC_V2_INTEGRATION_DECISION_V4.md) records five new consumed reservations and 32 output captures, separately from diff --git a/docs/README.md b/docs/README.md index c9e3f97..e0cc11b 100644 --- a/docs/README.md +++ b/docs/README.md @@ -4,9 +4,11 @@ records the single 2026-10-17 namespace failure: EXPERIMENTAL / EGL_FX_PROBE_BLOCKED_BEFORE_HELPER / EGL_NOT_RUN / ALLOCATOR_NOT_ATTEMPTED. The historical READY_FOR_APPROVAL verdict preceded that consumed attempt. -The [task-limit protocol design](design/STEP20C_EGL_FX_TASK_LIMIT_PROTOCOL.md) -selects probe-local cgroup admission for future implementation; -EGL_FX_RETRY_PREP_BLOCKED, new runtime NOT_RUN. No retry is authorized. +The [task-limit protocol](design/STEP20C_EGL_FX_TASK_LIMIT_PROTOCOL.md) +records probe-local cgroup admission as EXPERIMENTAL / IMPLEMENTED_OFFLINE / +OFFLINE_REVIEW_ACCEPTED, with inert validation only. Qualification against the real +manager and new EGL runtime remain NOT_RUN. Separate approval is required for +benign qualification, then for any new EGL attempt; no retry is authorized. [Gate 2 headless boot](design/STEP20C_HEADLESS_BOOT_GATE2.md) records the single V5 native failure: EXPERIMENTAL / PRISTINE_ALLOCATOR_BLOCKED / CORE_NOT_RUN, diff --git a/docs/ROADMAP.md b/docs/ROADMAP.md index ba9da04..601bce6 100644 --- a/docs/ROADMAP.md +++ b/docs/ROADMAP.md @@ -2,10 +2,14 @@ The [EGL/FX attempt](design/STEP20C_HEADLESS_EGL_FX_PROBE.md#observed-single-attempt--2026-10-17) failed at Bubblewrap namespace creation before the helper: EGL_NOT_RUN / -ALLOCATOR_NOT_ATTEMPTED; budget consumed. A probe-local cgroup task-limit -[protocol is DESIGN_ONLY](design/STEP20C_EGL_FX_TASK_LIMIT_PROTOCOL.md), -EGL_FX_RETRY_PREP_BLOCKED; new runtime NOT_RUN. No production contract or -STEP20C status changes, and no new native execution is authorized. +ALLOCATOR_NOT_ATTEMPTED; budget consumed. The probe-local cgroup task-limit +[protocol is IMPLEMENTED_OFFLINE](design/STEP20C_EGL_FX_TASK_LIMIT_PROTOCOL.md), +EXPERIMENTAL / OFFLINE_REVIEW_ACCEPTED, with inert validation only. The next gate is +new conversational approval of the reviewed benign qualification manifests, +before requesting separate benign qualification approval. Real manager +qualification and new EGL runtime remain NOT_RUN; a later EGL attempt needs +its own approval. No production contract or STEP20C status changes, and no new +native execution is authorized. The [CORE-C V2 follow-up V4](design/STEP20C_COREC_V2_INTEGRATION_DECISION_V4.md) is PARTIALLY_ADVANCED: five consumed runs/32 output captures, sampled migration/ diff --git a/docs/design/STEP20C_EGL_FX_TASK_LIMIT_PROTOCOL.md b/docs/design/STEP20C_EGL_FX_TASK_LIMIT_PROTOCOL.md index 949858e..e7eb350 100644 --- a/docs/design/STEP20C_EGL_FX_TASK_LIMIT_PROTOCOL.md +++ b/docs/design/STEP20C_EGL_FX_TASK_LIMIT_PROTOCOL.md @@ -1,12 +1,30 @@ # STEP20C — EGL/FX task-limit protocol design — 2026-10-17 -Status: **EXPERIMENTAL / DESIGN_ONLY / EGL_FX_RETRY_PREP_BLOCKED**. +Status: **EXPERIMENTAL / IMPLEMENTED_OFFLINE / OFFLINE_REVIEW_ACCEPTED**. New runtime: **NOT_RUN**. No scope, Bubblewrap, EGL/FX helper or compositor is -launched by this mission. This documentary batch deliberately leaves the -historical fixture source unchanged. It does not implement cgroup admission. +launched by this mission. The source-only runner now implements the gate; +qualification against the real manager remains separately approval-gated. The [observed failure](STEP20C_HEADLESS_EGL_FX_PROBE.md#observed-single-attempt--2026-10-17) precedes this design; its consumed reservation remains immutable. +## Adopted implementation slice — 2026-10-17 + +The implementation ticket authorizes source-only Python and inert tests now. +Use an explicitly owned transient user scope under app.slice, Delegate=yes, +TasksMax=256, with supervisor outside the scope and bounded manager runtime +and stop deadlines. A fixed bootstrap must wait before any payload exec; +validate the manager/kernel/process identities and release once only. +Channel EOF, invalid nonce, missing proof or timeout must fail closed. Keep +NPROC inherited, removing only the explicit 256 override after cgroup admission. +No real scope or payload may run in this implementation mission. Code and inert +behavioral validation establish IMPLEMENTED_OFFLINE; scoped code review is +accepted; final audit and its one bounded repair are complete. The +private observer repair has independent targeted acceptance. +Fresh private qualification LABs/manifests and exact benign commands are required; +benign qualification and EGL execution each need separate conversational approval. +The historical validation section below retains its original provenance and +does not describe current implementation coverage. + ## Evidence, guarantees and admission | Route / evidence | Risk | Bound actually supplied | Admission condition / disposition | @@ -14,7 +32,7 @@ precedes this design; its consumed reservation remains immutable. | Historical RLIMIT_NPROC=256 before bwrap; incident reports 1126 UID threads | Immediate task creation failure, also other possible EAGAIN causes | Real-UID thread accounting, not a probe-local quota | REJECT for retry; no syscall-level causal proof | | Raise NPROC from a procfs count | Count is racy; unrelated UID load changes; no descendant aggregate | Still only real-UID accounting | REJECT as replacement for task containment; no arbitrary count-plus-margin | | Omit explicit NPROC override, retain inherited soft/hard limits | Removes the intended extra UID restriction; inherited limit can still cause refusal | No new local task bound by itself | Only with independently verified cgroup admission; never silently raise inherited limits | -| Fresh delegated cgroup v2 with pids.max, possibly manager-owned transient user scope | Wrong subtree, post-spawn migration race, insufficient ancestor budget, unauthorized cleanup | Hierarchical task creation limit for admitted descendants, counting threads | SELECTED design direction, NOT IMPLEMENTED or locally qualified; all gates below required | +| Fresh delegated cgroup v2 with pids.max, possibly manager-owned transient user scope | Wrong subtree, post-spawn migration race, insufficient ancestor budget, unauthorized cleanup | Hierarchical task creation limit for admitted descendants, counting threads | IMPLEMENTED_OFFLINE; real qualification NOT_RUN; all gates below required | | Prior benign-child scope experiments in Gate 2 | Different launcher context, identity and lifecycle | Evidence for that earlier scope only | Do not inherit PASS; repeat only under new explicit approval | | AS 1 GiB / CPU 5 s / NOFILE 64 / FSIZE 1 MiB / CORE 0 | Multiple processes multiply several limits; tmpfs and total storage remain separate | Per-process address space/CPU/FD limits, per-file size, disabled core dumps | Preserve exact settings; no claim of aggregate memory/CPU/storage quota | | 8 s payload + 2 s cleanup; 1 MiB per raw pipe | Stuck cleanup, leaked descendants, parent death | Supervised deadline and captured-output bounds on observed paths | Preserve; failure to prove cleanup is BLOCKED, never a retry | @@ -29,7 +47,7 @@ counts TIDs hierarchically and rejects task creation over the limit. Migration can exceed pids.max, so a verified empty private group and controlled admission are essential. A namespace alone supplies no task quota. -## Minimal selected protocol, pending implementation and review +## Implemented protocol — real qualification pending Use one fresh, exclusively owned **domain** cgroup for the probe descendants; keep the collecting supervisor outside the kill target. Prefer the existing @@ -39,8 +57,8 @@ do not mkdir or write arbitrary manager-owned subtrees. A transient scope is only a candidate integration mechanism, not proof that its tasks were admitted before execution. No systemd-run or control-plane mutation is allowed now. -A future source-only patch is conditional on independent design acceptance. -It must implement the following invariants together, not merely delete NPROC: +The source-only runner implements the following invariants together. +Independent code review and native qualification are separate gates: 1. Validate cgroup2 mount identity from mountinfo, domain type, delegation, pids controller availability/enabling, exact owned path, directory identity, @@ -88,11 +106,11 @@ not an implicit expansion here. No global UID or manager configuration changes. ## Offline acceptance matrix — required before new preparation -These are **PLANNED / NOT IMPLEMENTED / NOT TESTED for the new protocol**. -The existing fixture's tests cover only its historical protocol. New tests -must call the actual new admission/refusal paths using fake procfs, cgroupfs, -manager and process adapters; no native scope, bwrap, helper or device call. -An execution sentinel must fail the test on any unintended native boundary. +These paths are exercised **OFFLINE** through the actual admission, +bootstrap, collection and cleanup functions, using fake procfs, cgroupfs, +manager/process adapters and local inert channels. The new test module installs +a native-spawn sentinel. Existing historical tests are adapted to schema2. +No real scope, bwrap, helper or device is involved; runtime remains NOT_RUN. | Offline case | Required observation | | --- | --- | @@ -115,28 +133,59 @@ patch or public ABI change is needed or authorized by this documentary design. ## Future authorization sequence — not executable instructions -First complete the conditional source-only implementation and offline review. -Then request a separate, bounded **benign-child cgroup qualification** approval: -new private 0700 LAB, exact sealed argv/environment, finite task-limit test, -admission-before-exec and timeout/parent-death cleanup evidence. Historical -scope experiments do not waive this gate. No EGL or Bubblewrap in that phase. +The runner now has separate schema2 benign/EGL kinds and approval flags. +`prepare.py --benign --qualification-mode nominal|parent-death` creates fresh +0700 qualification LABs and exact sealed passive commands. Independent review +must accept the final scripts and manifests before qualification approval. +The nominal payload is fixed Python; the distinct parent-death LAB deliberately +exits its supervisor after release, with manager runtime/stop limits as the +independent cleanup authority. Its result requires external observation; +a supervisor that has exited cannot certify its own cleanup or write CLEANED. +Its private gate.sock may remain as an artifact. External observation must be +read-only and bind the same invocation, cgroup identity and task lifetimes; +absence of that evidence is not successful cleanup. -Only after that qualification passes, prepare a distinct fresh 0700 EGL/FX LAB -and revised manifest with the new launcher, effective resource policy, -identities, exact command, ownership and cleanup protocol. Preserve pinned -sources/library ABI and the existing closed device/mount/FD policy. Obtain -independent review on that exact manifest and request **new conversational -approval** for one EGL/FX attempt with its own budget. Do not spend it during -preparation, inherit the old approval, or silently retry any failure. +After benign qualification passes, prepare **another** fresh 0700 EGL LAB and +manifest. Preserve exact source/library ABI and the closed mount/device/FD policy; +obtain review and a separate new conversational EGL approval. No historical +reservation or approval is reusable. The old NPROC256 fixture remains preserved +in its immutable historical LAB; repository schema2 accepts only the new policy. +No EGL retry is authorized by source-only readiness or by benign qualification. -There is deliberately **no exact future native command** in this edition: -no implemented/qualified cgroup launcher or new reviewed manifest exists. -**EGL_FX_RETRY_PREP_BLOCKED** identifies those preparation gates, not a new EGL -failure. It does not require a human decision to weaken containment. The next -smallest technical step is reviewed source-only admission implementation; -any later native qualification still needs explicit user authorization. +## Implementation evidence — 2026-10-17 -## Validation of this documentary batch — 2026-10-17 +`cgroup_gate.py` implements read-only preflight, explicit manager-owned scope +creation, AF_UNIX SOCK_SEQPACKET admission with SO_PEERCRED on both ends, +PID/startticks/pidfd and invocation/inode checks, single release, bounded drain +and owned cleanup. A bounded observer reads recursive emptiness concurrently; +ENOENT/ENODEV is never a substitute for populated=0. Missing evidence produces +CLEANUP_UNPROVEN. The bootstrap closes descriptors above stderr before exec. + +The helper is loaded only after its digest check and without bytecode writes; +real isolated passive-CLI tests exercise the no-write boundary. Scope properties +are TasksMax=256, RuntimeMaxSec=12s, TimeoutStopSec=2s, KillMode=control-group: +the manager deadline bounds orphaned execution independently of the supervisor; +normal supervision uses a 10-second deadline including cleanup. These limits +remain local experimental values to qualify, not measured Mesa requirements. +New manifests seal the scope command/properties, scripts, tool ELF identities +and dependency closure. No C, compositor, ABI or packaging change is made. + +The inert suite passes; exact final counts and source bindings belong to the +final review and private handoff after the bounded repairs. A real +passive preparation exposed a systemctl RUNPATH resolution defect; the failed +fresh LAB and traceback are retained privately and the defect is repaired with +explicit ELF resolution tests before final preparation. No native attempt was +consumed by that preparation failure. + +Pre-admission manager diagnostics are not retained as raw stdout/stderr; the +supervisor error is retained. This is a bounded diagnostic limitation, not +proof of manager behavior. Neither offline tests nor the manager deadline +establish actual cleanup after supervisor death: nominal and parent-death +qualification remain NOT_RUN. Final review may establish readiness to request +benign approval; this edition does not predeclare that readiness or authorize +any scope, benign payload, Bubblewrap or EGL execution. + +## Historical validation of the preceding documentary batch — 2026-10-17 The existing `test_probe.py` suite passes **23/23** with `PYTHONDONTWRITEBYTECODE=1 python3 -m unittest discover -s tooling/probes/step20c-egl-fx -p test_probe.py -v`. @@ -154,3 +203,31 @@ inventory, log, private manifest, binary or personal scope identifier is publish Native build, C harness, cgroup qualification, Bubblewrap and EGL/FX: **NOT RUN** in this mission. Independent review and actual publication outcome are reported in the task handoff; this document does not predeclare their PASS. + +## Source-only closeout — 2026-10-17 + +**CGROUP_PROBE_IMPLEMENTED_OFFLINE_READY_FOR_BENIGN_APPROVAL**. +Repository behavior tests: **75 PASS**; Python AST5, local link targets and +diff whitespace checks pass. Two fresh benign LABs contain matching source +fixtures,47 sealed file identities and39 ELF records each; their exact isolated +passive commands pass without writes or ATTEMPT. The actual native runner and +external observer remain **NOT_RUN**. + +The single final audit found one BLOCKING in the private parent-death observer: +it could accept cleanup without observing an executed benign payload surviving +its supervisor. The bounded repair requires exact cmdline/PID/startticks/cgroup, +same scope invocation/inode/policy, at least1s observed survival after the +supervisor lifetime disappears, then observed recursive emptiness and unit/cgroup +absence within14s (global30s). Fourteen deterministic tests pass; independent +targeted review accepts this repair. No second general final audit was run. +The configuration alone is not claimed to prove RuntimeMax caused termination. + +Private LABs retain REPORT_IMPLEMENTATION.md, BENIGN_GATE_APPROVAL.md and +EGL_GATE_NEXT.md, reviewed source/diff hashes and test evidence. Nominal and +parent-death cases have distinct budgets; the latter needs its concurrent +read-only observer. Exact private paths/digests belong to the approval dossier +and task handoff, not public unit identifiers. Historical evidence and consumed +reservations are unchanged. Real manager admission, parent-death cleanup and +Mesa task sufficiency remain unqualified. Separate conversational approval is +required before benign acquisition, then another fresh LAB/review/approval +before any EGL attempt. STEP20C remains NEXT, multi-output NOT_YET_SUPPORTED. diff --git a/docs/design/STEP20C_HEADLESS_EGL_FX_PROBE.md b/docs/design/STEP20C_HEADLESS_EGL_FX_PROBE.md index 5afe8c3..df25b23 100644 --- a/docs/design/STEP20C_HEADLESS_EGL_FX_PROBE.md +++ b/docs/design/STEP20C_HEADLESS_EGL_FX_PROBE.md @@ -2,7 +2,8 @@ Status: **EXPERIMENTAL / EGL_FX_PROBE_BLOCKED_BEFORE_HELPER / EGL_NOT_RUN / ALLOCATOR_NOT_ATTEMPTED**. The single historical launch budget is consumed; no retry is authorized. -Retry preparation: **EGL_FX_RETRY_PREP_BLOCKED / runtime NOT_RUN**. +New admission implementation: **EXPERIMENTAL / IMPLEMENTED_OFFLINE / OFFLINE_REVIEW_ACCEPTED**; +real cgroup qualification and new EGL runtime **NOT_RUN**. See the [task-limit protocol](STEP20C_EGL_FX_TASK_LIMIT_PROTOCOL.md). The source-only fixture prepares one future measurement, without authorizing execution. STEP20C remains NEXT; production multi-output NOT_YET_SUPPORTED. @@ -62,8 +63,9 @@ ABI or a mandatory host device blocks preparation instead of weakening isolation ## Source-only preparation and identity boundary -The [six-file fixture](../../tooling/probes/step20c-egl-fx/README.md) contains -probe.c, prepare.py, run.py, test_probe.py, test_probe_mocks.c and README.md. +The [eight-file fixture](../../tooling/probes/step20c-egl-fx/README.md) contains +probe.c, prepare.py, run.py, cgroup_gate.py, test_probe.py, +test_cgroup_admission.py, test_probe_mocks.c and README.md. Its README gives exact preparation and inert-test commands. prepare.py creates a new private mode0700 LAB, extracts SceneFX and wlroots from pinned Git objects, checks every extracted blob, compares compiled installed headers and @@ -74,8 +76,8 @@ neither the target loader, ldd nor the real helper is executed. The private manifest seals argv, environments, compile commands, cc -M inputs, source/header/script/tool/helper hashes, ELF identities and DT_NEEDED closure. -The prepared manifest covers 755 files, 80 ELF identities, 17 compared installed -headers and 557 source blobs. Installed ELF bytes/Build IDs are pinned, not +The historical prepared manifest covers 755 files, 80 ELF identities, +17 compared installed headers and 557 source blobs. Installed ELF bytes/Build IDs are pinned, not reproduced from source; header/export agreement proves no binary source identity. Mesa EGL, swrast, GBM dri and gallium providers are explicit closure seeds; conditional runtime dlopen remains unobserved. The inspected ELF closure has @@ -157,7 +159,7 @@ Read-only inspection in this mission confirms ATTEMPT with budget_consumed=1, stdout length 0, stderr length 71, and result cause=exit, returncode=1, collected=true, elapsed=0.003128459007712081 seconds, error=null. The diagnostic is `bwrap: Creating new namespace failed: Resource temporarily unavailable`. -The retained launcher is byte-identical to repository run.py and applies +The retained launcher is byte-identical to run.py at baseline `99f1fbeb` and applies `prlimit --nproc=256` before Bubblewrap creates namespaces. The namespace creation failed before the C helper, EGL or SceneFX started. @@ -180,5 +182,30 @@ preparation evidence, with no new runtime success. The reservation is exhausted even though the helper never started. Never remove ATTEMPT, modify the old LAB, rerun its launcher, or reuse its digest to admit revised inputs. No new native process, Bubblewrap, scope, device exposure or compositor was authorized or -executed in this follow-up. The [new protocol design](STEP20C_EGL_FX_TASK_LIMIT_PROTOCOL.md) -requires separately reviewed implementation and new conversational approval. +executed in this incident follow-up. The subsequent +[task-limit protocol](STEP20C_EGL_FX_TASK_LIMIT_PROTOCOL.md) now records the +source-only implementation separately from this historical result. + +## Subsequent cgroup admission implementation — offline only + +The schema2 runner implements a manager-owned probe-local scope, a blocked +AF_UNIX peer-credential admission handshake before payload exec, inherited +NPROC without the explicit 256 override, bounded collection and owned cleanup. +The scope budget is TasksMax=256, RuntimeMaxSec=12s and TimeoutStopSec=2s; +normal supervision remains bounded to 10 s including cleanup. Mocked/inert +validation exercises the admission and refusal paths without a real scope, +Bubblewrap or EGL helper. This establishes IMPLEMENTED_OFFLINE, with final +review accepted after a bounded private-observer repair; it does not +qualify real manager behavior or Mesa capacity. + +Fresh benign nominal and parent-death LABs require distinct manifests and budgets. +Parent-death qualification requires external read-only observation: a dead +supervisor cannot certify CLEANED, and its private gate.sock can remain as an +artifact. Pre-admission manager diagnostics are not retained as raw output; +the supervisor error remains available. These limits must stay explicit in +the qualification handoff. Real qualification is NOT_RUN and requires new +conversational approval after the accepted offline review. Only after benign qualification +may a separately reviewed fresh EGL LAB be proposed for its own approval. +The historical ATTEMPT, manifest, failure and exhausted one-launch budget +remain unchanged. STEP20C stays NEXT; production multi-output remains +NOT_YET_SUPPORTED. diff --git a/tooling/probes/step20c-egl-fx/README.md b/tooling/probes/step20c-egl-fx/README.md index d3ae19a..4549b91 100644 --- a/tooling/probes/step20c-egl-fx/README.md +++ b/tooling/probes/step20c-egl-fx/README.md @@ -1,19 +1,18 @@ # STEP20C — fixture source-only EGL/FX -**Mise à jour du 17/10/2026 : essai unique consommé, BLOCKED_BEFORE_HELPER.** -Bubblewrap a échoué à créer le namespace ; EGL_NOT_RUN / -ALLOCATOR_NOT_ATTEMPTED. Les instructions ci-dessous décrivent la préparation -historique et ne constituent plus une procédure de relance admissible. -Ne relancer aucun ancien LAB. Le [nouveau protocole de limite de tâches](../../../docs/design/STEP20C_EGL_FX_TASK_LIMIT_PROTOCOL.md) -est DESIGN_ONLY / EGL_FX_RETRY_PREP_BLOCKED : admission cgroup à implémenter, -tester hors ligne et qualifier séparément avant toute nouvelle autorisation EGL. -Le code historique reste inchangé ; ses tests ne valident pas ce nouveau protocole. +**EXPERIMENTAL — admission cgroup implémentée pour validation hors ligne ; +qualification du nouveau runner NOT_RUN.** L'ancien essai reste consommé, +BLOCKED_BEFORE_HELPER, EGL_NOT_RUN / ALLOCATOR_NOT_ATTEMPTED. Aucun ancien LAB +ne doit être modifié ou relancé. Cette fixture ne change ni C, ni compositeur, +ni ABI, ni packaging. STEP20C reste NEXT et le clipping multi-output de +production NOT_YET_SUPPORTED. -**EXPERIMENTAL / runtime NOT_RUN.** Cette fixture prépare une unique mesure -future. Elle n'ajoute aucune dépendance de production, ne démarre aucun -compositeur et ne démontre ni allocation ni rendu. Le document de référence est -[la proposition de sonde](../../../docs/design/STEP20C_HEADLESS_EGL_FX_PROBE.md). -Les anciens LAB restent en lecture seule. +Le [contrat d'admission](../../../docs/design/STEP20C_EGL_FX_TASK_LIMIT_PROTOCOL.md) +et [l'incident EGL](../../../docs/design/STEP20C_HEADLESS_EGL_FX_PROBE.md) +distinguent les preuves historiques de cette implémentation. Les tests mockés +ne prouvent aucun lancement sur le manager réel et ne valident pas le budget +Mesa. L'exécution bénigne et une éventuelle exécution EGL demandent chacune une +nouvelle autorisation conversationnelle avec leur propre LAB et manifeste. `probe.c` crée seulement un event loop, un backend headless non démarré et EGL via le constructeur interne pinned `wlr_egl_create_with_drm_fd(-1)`. Il interroge @@ -54,12 +53,15 @@ ne sont **pas reproduites depuis les sources**, et la correspondance des headers/exportations ne prouve pas leur provenance source. Les licences restent dans les archives privées ; aucun binaire, log ou manifeste privé n'entre dans Git. -Les tests Python ne lancent que des processus Python inertes pour exercer les -pipes, le timeout et la collecte. Aucun test n'invoque bwrap ou le helper : +Les 23 tests historiques lancent seulement des enfants Python inertes pour +les pipes. `test_cgroup_admission.py` exerce les adaptateurs et l'orchestrateur +réels avec faux manager/proc/cgroup/processus, sockets locaux et threads inertes. +Une sentinelle interdit toute création native dans cette nouvelle suite. +Aucun test n'invoque systemd-run, bwrap ou le helper : ```sh PYTHONDONTWRITEBYTECODE=1 python3 -m unittest discover \ - -s tooling/probes/step20c-egl-fx -p test_probe.py -v + -s tooling/probes/step20c-egl-fx -p 'test_*.py' -v ``` Le harnais C substitue **tous** les appels EGL/Wayland/wlroots et inclut le @@ -104,15 +106,87 @@ ce choix ne prouve pas l'identité logicielle, vérifiée séparément par EGL. Mesa EGL, swrast, GBM dri et gallium plus leurs DT_NEEDED sont scellés ; des chargements conditionnels runtime restent non observés jusqu'à l'unique essai. -Le budget est exactement un lancement, 8 s de payload et 2 s de réserve de -nettoyage dans un délai nominal supervisé de 10 s au total, 1 MiB par canal brut, AS 1 GiB, CPU 5 s, -NOFILE 64, FSIZE 1 MiB, CORE 0, NPROC 256. AS/CPU sont par processus et NPROC -s'applique à l'UID global ; ce n'est pas une limite agrégée cgroup. Le -superviseur crée sa session, ferme les FDs, lit les deux pipes concurremment, -utilise pidfd si disponible, tue seulement son groupe non récolté et récolte -son enfant. La fin du namespace PID bwrap termine ses descendants. Une -collecte non prouvée est un échec explicite ; aucun PID déjà récolté n'est tué. -Ce contrat ne prétend pas durcir le système contre un pair hostile du même UID, -ni retirer les capacités de diagnostic `/proc` déjà imposées au protocole. -Les canaux bruts et reçus bornés restent privés, avec écritures atomiques et -fsync ; aucun scope systemd n'est créé. +La préparation bénigne inspecte passivement les outils Python/systemd/env/readelf +avec readelf : Build ID, ELF64 x86-64, interpréteur fermé et fermeture DT_NEEDED +sont enregistrés ; tous les octets de cette fermeture sont scellés. Elle ne +compile ni ne charge EGL et ne demande pas de source-store. + +Le manifeste schema2 scelle `scope_argv` (seuls PID/startticks du superviseur +restent des emplacements remplis par préflight), `kind` (`benign` ou `egl`), `qualification_mode`, +la commande fermée, les environnements et `scope` : nonce aléatoire 128 bits, +nom et Description associés, Delegate=yes, TasksMax=256, RuntimeMaxSec=12s, +TimeoutStopSec=2s et KillMode=control-group. Le timeout manager indépendant +borne les descendants en cas de disparition du superviseur ; cette propriété +reste à qualifier sur le vrai manager. La limite cible `pids.max=256` compte +tous les threads du sous-arbre, sans garantie de suffisance Mesa. + +`run.py` sans drapeau d'approbation vérifie seulement les identités et imprime +le plan : aucun processus, fichier ATTEMPT, socket ni appel manager. Le helper +Python est chargé par chemin absolu, après vérification de son hash, même avec +`python3 -I`. Avec approbation, le préflight refuse un vrai launcher en session +scope, un manager non délégué, app.slice inactif, un montage non cgroup2/domain, +un ancêtre non writable, pids absent, un headroom insuffisant, un diagnostic +UID incertain ou NPROC hérité sous la charge UID. Aucun snapshot /proc ne +constitue une réservation. Aucun processus n'est migré par le superviseur. + +Après ATTEMPT durable O_EXCL, systemd-run crée un seul scope et lance le +bootstrap fixe. Celui-ci ouvre un socket AF_UNIX SOCK_SEQPACKET privé mode0600 +dans le LAB0700, sans hypothèse de transmission stdin/FD par systemd-run. +Les deux côtés vérifient SO_PEERCRED ; le bootstrap vérifie le startticks et +le pidfd du superviseur. Avant RELEASE unique, le superviseur contrôle la +Description nonce, Id/InvocationID/ControlGroup, propriétés manager, inode et +propriétaire cgroup, pids.max effectif256, pids.current1, unique PID, absence +de sous-cgroup, UID/startticks/pidfd et membership. EOF, nonce faux, timeout, +PID mort ou divergence interdisent l'exec. Aucun prlimit/bwrap/EGL ne précède +RELEASE. Les FD supérieurs à2 sont fermés à la frontière exec ; le bus manager +ne rejoint pas la charge et l'environnement devient exactement HOST_ENV. + +Le payload conserve AS1GiB, CPU5s, NOFILE64, FSIZE1MiB et CORE0. Seul +l'override NPROC256 historique est supprimé ; les limites soft/hard NPROC sont +héritées, jamais augmentées. AS/CPU/NOFILE sont des limites par processus, +FSIZE est par fichier ; aucun quota agrégé mémoire/CPU/disque n'est promis. +Le délai du runner est 10s depuis préflight (collecte jusqu'à8s, cleanup +jusqu'à10s), avec 1MiB par pipe. Le filet manager12s+stop2s est séparé. + +Un observateur inerte commence avant RELEASE et échantillonne l'inode et +`cgroup.events populated=0`, `cgroup.procs`, `cgroup.threads`, `pids.current` +récursivement. Il reste actif pendant collecte/stop. Le cleanup vérifie la +même invocation/inode avant tout stop. Un scope déjà disparu n'est accepté +qu'avec une preuve vide antérieure et la disparition unité+cgroup. Une course +qui cache l'état vide reste CLEANUP_UNPROVEN, sans convertir ENOENT/ENODEV en +preuve. Ce protocole est strict : la qualification réelle doit mesurer si +l'échantillonnage observe effectivement la transition. Aucun stop par nom seul, +aucun PID découvert n'est tué, aucun repli vers pkill/cgroup.kill. + +Les phases privées JSON sont PRECHECK, SCOPE_CREATED (requête de création), +ADMITTED (ownership), RELEASED (intention exec), COLLECTED et CLEANED. +PAYLOAD_STARTED requiert le marqueur stdout fixe du seul enfant bénin ; une +intention exec n'est pas une preuve d'exécution EGL. Les sorties EGL restent +brutes et `EGL_RESULT_UNINTERPRETED`. Les préfixes de sortie sont conservés +sur overflow/timeout. Toute collecte/cleanup/journalisation non prouvée refuse +le succès. Le risque d'interférence hostile du même UID n'est pas éliminé. + +Préparer deux LABs **neufs distincts**, sans compilation ni exécution native : + +```sh +python3 tooling/probes/step20c-egl-fx/prepare.py \ + --lab /var/tmp/NOUVEAU_LAB_BENIN_NOMINAL --benign --qualification-mode nominal +python3 tooling/probes/step20c-egl-fx/prepare.py \ + --lab /var/tmp/NOUVEAU_LAB_BENIN_PARENT --benign --qualification-mode parent-death +``` + +Le cas nominal affiche un marqueur et dort0.2s. Le cas parent-death affiche +le même marqueur et dort30s ; après RELEASE et reçu durable PARENT_DEATH_TEST, +le superviseur termine volontairement avec code71. Le manager doit alors +arrêter le scope à12s, avec stop2s. Le résultat de ce cas doit être établi par +un observateur extérieur ; le superviseur disparu ne peut écrire son succès. +L'observateur doit vérifier la même Description/InvocationID/inode, la fin des +tâches, l'état vide et la disparition, sans jamais stopper une unité étrangère. +Les commandes exactes avec paths/digests/noms éphémères et l'observation bornée +appartiennent aux documents d'autorisation privés du nouveau LAB. + +Après accord distinct seulement, depuis le launcher app.slice prévu, utiliser +la commande passive imprimée avec `--approve-benign-gate`. Ce drapeau est +incompatible avec un manifeste EGL ; `--approve-native-probe` est incompatible +avec un manifeste bénin. Aucun scope réel, charge bénigne, bwrap ou EGL n'a été +exécuté par l'implémentation et les tests hors ligne. diff --git a/tooling/probes/step20c-egl-fx/cgroup_gate.py b/tooling/probes/step20c-egl-fx/cgroup_gate.py new file mode 100644 index 0000000..8b7c7fd --- /dev/null +++ b/tooling/probes/step20c-egl-fx/cgroup_gate.py @@ -0,0 +1,724 @@ +#!/usr/bin/env python3 +"""Admission expérimentale cgroup v2 ; aucun effet à l'importation. + +Le seul bootstrap natif est fixe et bloqué sur un socket privé. systemd-run +n'a pas besoin de transmettre un descripteur : le bootstrap se connecte par +nom, puis les deux extrémités vérifient SO_PEERCRED et la durée de vie PID. +""" +import argparse +import json +import os +from pathlib import Path +import re +import resource +import select +import selectors +import socket +import stat +import struct +import subprocess +import sys +import threading +import time + +CLEAN_ENV = {"PATH": "/usr/bin", "LC_ALL": "C"} +TASKS = 256 +# systemctl show renders these D-Bus microsecond properties as durations. +# Compare its exact CLI representation, matching the fixed Sec= launch options. +POLICY = {"Delegate": "yes", "TasksMax": "256", "RuntimeMaxUSec": "12s", + "TimeoutStopUSec": "2s", "KillMode": "control-group"} +LIMIT = 1048576 + + +class Refusal(RuntimeError): + """Une preuve absente ou contradictoire interdit la libération.""" + + +def require(condition, message): + if not condition: + raise Refusal(message) + + +def policy(nonce): + """Identité fraîche, scellée dans le manifeste ; jamais un nom réutilisable.""" + require(re.fullmatch(r"[0-9a-f]{32}", nonce) is not None, "nonce invalide") + return {"nonce": nonce, "unit": "mosaic-egl-" + nonce + ".scope", + "description": "Mosaic experimental admission " + nonce, + "properties": dict(POLICY), "pids_max": TASKS} + + +def scope_command(spec, bootstrap_argv): + """Construction fermée du mécanisme manager ; aucune option libre.""" + argv = ["/usr/bin/systemd-run", "--user", "--scope", "--slice=app.slice", + "--collect", "--no-ask-password", "--expand-environment=no", + "--unit=" + spec["unit"], "--description=" + spec["description"]] + properties = ["Delegate=yes", "TasksMax=256", "RuntimeMaxSec=12s", + "TimeoutStopSec=2s", "KillMode=control-group"] + return argv + ["--property=" + value for value in properties] + bootstrap_argv + + +def bootstrap_command(lab, spec, identity, argv): + return ["/usr/bin/python3", "-I", str(lab / "fixture/cgroup_gate.py"), + "--bootstrap", str(lab / "gate.sock"), spec["nonce"], + str(identity["pid"]), str(identity["startticks"]), *argv] + + +def sealed_scope_argv(lab, manifest): + """Seuls PID/startticks seront remplis par le préflight de cette invocation.""" + identity = {"pid": "", "startticks": ""} + return scope_command(manifest["scope"], bootstrap_command( + lab, manifest["scope"], identity, manifest["argv"])) + + +class ProcFS: + """Observations procfs strictes, sans migration ni changement de rlimit.""" + + def __init__(self, root=Path("/proc")): + self.root = root + + def identity(self, pid): + directory = self.root / str(pid) + before = directory.stat() + fields = (directory / "stat").read_text().rsplit(")", 1)[1].split() + lines = (directory / "cgroup").read_text().splitlines() + require(len(lines) == 1 and lines[0].startswith("0::/"), "cgroup v1/hybride") + status = dict(line.split(":", 1) for line in + (directory / "status").read_text().splitlines() if ":" in line) + require(set(map(int, status["Uid"].split())) == {before.st_uid}, + "UID process divergent") + require(directory.stat().st_ino == before.st_ino, "proc identity drift") + return {"pid": pid, "uid": before.st_uid, "startticks": int(fields[19]), + "cgroup": lines[0][3:]} + + def uid_threads(self, uid): + total = 0 + # Every disappearing/unreadable process refuses; a snapshot never reserves. + for directory in self.root.iterdir(): + if not directory.name.isdecimal(): + continue + status = dict(line.split(":", 1) for line in + (directory / "status").read_text().splitlines() + if ":" in line) + observed_uid = int(status["Uid"].split()[0]) + if observed_uid == uid: + before = self.identity(int(directory.name)) + require(before["uid"] == uid, "UID drift pendant diagnostic") + threads = int(status["Threads"]) + tasks = list((directory / "task").iterdir()) + require(threads > 0 and threads == len(tasks), + "compte threads incohérent") + after_status = dict(line.split(":", 1) for line in + (directory / "status").read_text().splitlines() + if ":" in line) + require(status["Uid"] == after_status["Uid"] + and status["Threads"] == after_status["Threads"] + and self.identity(int(directory.name)) == before, + "UID/count/startticks drift pendant diagnostic") + total += threads + require(total > 0, "charge UID inconnue") + return total + + def nproc(self): + return resource.getrlimit(resource.RLIMIT_NPROC) + + def pidfd(self, pid): + return os.pidfd_open(pid) + + +class CgroupFS: + """Accès exclusivement en lecture au montage unifié et à ses compteurs.""" + + def __init__(self, root=Path("/sys/fs/cgroup")): + self.root = root + + def mount(self): + matches = [] + for line in Path("/proc/self/mountinfo").read_text().splitlines(): + left, right = line.split(" - ", 1) + fields = left.split() + if fields[4] == str(self.root): + require(right.split()[0] == "cgroup2" and fields[3] == "/", + "montage cgroup2 unifié racine requis") + matches.append((int(fields[0]), fields[2])) + require(len(matches) == 1, "identité montage ambiguë") + return matches[0] + + def path(self, group): + require(group.startswith("/") and ".." not in group.split("/"), + "chemin cgroup invalide") + path = self.root / group.lstrip("/") + require(path.resolve(strict=True) == path, "symlink cgroup refusé") + require(path.stat().st_dev == self.root.stat().st_dev, + "montage cgroup divergent") + return path + + def snapshot(self, group): + path = self.path(group) + for name in ("cgroup.type", "pids.max", "pids.current", "cgroup.procs"): + require(not (path / name).is_symlink(), "symlink fichier cgroup refusé") + require((path / "cgroup.type").read_text().strip() == "domain", + "cgroup non domain/threaded") + info = path.stat() + return {"inode": info.st_ino, "device": info.st_dev, "uid": info.st_uid, + "pids_max": (path / "pids.max").read_text().strip(), + "pids_current": int((path / "pids.current").read_text()), + "procs": [int(pid) for pid in + (path / "cgroup.procs").read_text().split()], + "children": [entry.name for entry in path.iterdir() if entry.is_dir()]} + + def available(self, group): + path = self.path(group) + return ((path / "cgroup.controllers").read_text().split(), + (path / "cgroup.subtree_control").read_text().split(), + os.access(path / "cgroup.procs", os.W_OK, effective_ids=True)) + + def exists(self, group): + path = self.root / group.lstrip("/") + return path.exists() or path.is_symlink() + + def empty(self, group): + path = self.path(group) + def fail_walk(error): + raise error + + walk = os.walk(path, followlinks=False, onerror=fail_walk) + for directory, names, files in walk: + current = Path(directory) + require(not any((current / name).is_symlink() for name in names + files), + "symlink dans sous-arbre") + events = dict(line.split() for line in + (current / "cgroup.events").read_text().splitlines()) + if events.get("populated") != "0": + return False + if (current / "cgroup.procs").read_text().strip(): + return False + if (current / "cgroup.threads").read_text().strip(): + return False + if int((current / "pids.current").read_text()) != 0: + return False + return True + + +class DrainFailure(Refusal): + """Une collecte échouée conserve les préfixes bruts déjà observés.""" + + def __init__(self, message, channels): + super().__init__(message) + self.channels = [bytes(channel) for channel in channels] + + +def drain(process, deadline, limit=LIMIT): + """Collecte bornée ; le propriétaire effectue le stop et le wait final.""" + channels = [bytearray(), bytearray()] + try: + with selectors.DefaultSelector() as selector: + for index, stream in enumerate((process.stdout, process.stderr)): + os.set_blocking(stream.fileno(), False) + selector.register(stream, selectors.EVENT_READ, index) + while selector.get_map(): + remaining = deadline - time.monotonic() + require(remaining > 0, "drain timeout") + for key, _ in selector.select(min(.05, remaining)): + chunk = os.read(key.fd, 65536) + if not chunk: + selector.unregister(key.fileobj) + continue + target = channels[key.data] + room = limit - len(target) + target.extend(chunk[:room]) + require(len(chunk) <= room, "stdout/stderr overflow") + process.wait(timeout=max(.001, deadline - time.monotonic())) + return channels + except BaseException as error: + raise DrainFailure(f"{type(error).__name__}: {error}", channels) from error + + +class SystemManager: + """Les seules mutations sont la création scellée et le stop possédé.""" + + def __init__(self, uid, popen=None): + self.popen = popen or subprocess.Popen + self.environment = { + **CLEAN_ENV, "XDG_RUNTIME_DIR": f"/run/user/{uid}", + "DBUS_SESSION_BUS_ADDRESS": f"unix:path=/run/user/{uid}/bus", + "SYSTEMD_PAGER": "cat", "SYSTEMD_COLORS": "0", + } + + def spawn(self, argv): + return self.popen(argv, stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, + stderr=subprocess.PIPE, close_fds=True, + env=self.environment, start_new_session=True) + + def call(self, argv, deadline): + """Auxiliaire borné, récolté même si la lecture échoue.""" + require(deadline - time.monotonic() > .1, "budget manager épuisé") + process = self.spawn(argv) + try: + channels = drain(process, deadline - .05, limit=65536) + require(process.returncode == 0, "manager commande refusée") + return bytes(channels[0]).decode("utf-8", errors="strict") + finally: + try: + if process.poll() is None: + process.kill() # Our unreaped Popen only; never a discovered PID. + process.wait(timeout=max(.001, deadline - time.monotonic())) + finally: + process.stdout.close() + process.stderr.close() + + def show(self, unit, deadline, system=False): + argv = ["/usr/bin/systemctl"] + ([] if system else ["--user"]) + text = self.call(argv + ["show", unit, "--no-pager"], deadline) + return dict(line.split("=", 1) for line in text.splitlines() if "=" in line) + + def stop(self, unit, deadline): + argv = ["/usr/bin/systemctl", "--user", "--no-ask-password", "stop", unit] + self.call(argv, deadline) + + def create(self, spec, bootstrap): + return self.spawn(scope_command(spec, bootstrap)) + + +def ancestors(group): + current = Path(group) + while str(current) != "/": + yield str(current) + current = current.parent + + +def capacity(cgroups, app): + """Le headroom observé est un critère de refus, pas une réservation.""" + records = {} + for group in ancestors(app): + record = cgroups.snapshot(group) + count = record["pids_current"] + require(count >= 0, "pids.current négatif") + maximum = record["pids_max"] + if maximum != "max": + require(maximum.isdecimal(), "pids.max ancêtre invalide") + require(int(maximum) - count >= TASKS, "capacité ancêtre insuffisante") + records[group] = record + return records + + +def preflight(proc, cgroups, manager, uid, pid, deadline): + """Toutes les observations précèdent ATTEMPT et la première mutation.""" + require(uid != 0 and uid == os.geteuid(), "UID réel/effectif invalide") + mount = cgroups.mount() + base = f"/user.slice/user-{uid}.slice/user@{uid}.service" + app = base + "/app.slice" + launcher = proc.identity(pid) + require(launcher["uid"] == uid and launcher["cgroup"].startswith(app + "/"), + "launcher hors app.slice ; session scope refusé") + owner = manager.show(f"user@{uid}.service", deadline, system=True) + require(owner.get("Id") == f"user@{uid}.service" + and owner.get("Delegate") == "yes" and owner.get("ActiveState") == "active" + and owner.get("ControlGroup") == base, "manager non délégué/invalide") + require(re.fullmatch(r"[0-9a-f]{32}", owner.get("InvocationID", "")), + "identité invocation manager absente") + manager_process = proc.identity(int(owner["MainPID"])) + require(manager_process["uid"] == uid + and manager_process["cgroup"].startswith(base + "/"), + "identité process manager divergente") + application = manager.show("app.slice", deadline) + require(application.get("Id") == "app.slice" + and application.get("ActiveState") == "active" + and application.get("ControlGroup") == app, "app.slice inactif/invalide") + controllers, enabled, writable = cgroups.available(app) + require("pids" in controllers and "pids" in enabled, "contrôleur pids absent") + require(writable, "ancêtre commun non writable") + limits = proc.nproc() + load = proc.uid_threads(uid) + require(all(value == resource.RLIM_INFINITY or value >= load for value in limits), + f"inherited RLIMIT_NPROC insuffisant : {limits}, charge UID={load}") + return {"launcher": launcher, "manager": owner, "manager_process": manager_process, + "mount": mount, "app": app, "uid": uid, "nproc": limits, + "uid_threads_diagnostic": load, "ancestors": capacity(cgroups, app)} + + +def same_lifetime(first, second): + return all(first[key] == second[key] for key in ("pid", "uid", "startticks")) + + +def scope_identity(spec, observed, group): + """Nonce Description + invocation + chemin ; aucun stop sur nom seul.""" + for key, expected in (("Id", spec["unit"]), ("Description", spec["description"]), + ("ControlGroup", group)): + require(observed.get(key) == expected, "scope identité divergente : " + key) + require(re.fullmatch(r"[0-9a-f]{32}", observed.get("InvocationID", "")), + "scope InvocationID absent") + + +class Barrier: + """Socket privé, authentifié et à libération unique, sans FD relayé.""" + + def __init__(self, path, nonce, uid, pid, deadline): + self.path = path + self.nonce = nonce + self.uid = uid + self.pid = pid + self.deadline = deadline + self.connection = None + self.released = False + require(len(os.fsencode(path)) < 104, "nom socket trop long") + self.listener = socket.socket(socket.AF_UNIX, socket.SOCK_SEQPACKET) + try: + self.listener.bind(str(path)) + os.chmod(path, 0o600) + self.listener.listen(1) + except BaseException: + self.listener.close() + raise + + def accept(self): + require(self.connection is None and not self.released, + "bootstrap replay refusé") + self.listener.settimeout(max(.001, self.deadline - time.monotonic())) + self.connection, _ = self.listener.accept() + self.connection.settimeout(max(.001, self.deadline - time.monotonic())) + pid, uid, _ = struct.unpack("3i", self.connection.getsockopt( + socket.SOL_SOCKET, socket.SO_PEERCRED, struct.calcsize("3i"))) + require(uid == self.uid, "bootstrap UID divergent") + message = json.loads(self.connection.recv(4096)) + require(message == {"nonce": self.nonce, "pid": pid}, + "bootstrap nonce/PID invalide") + return pid + + def release(self): + require(not self.released, "seconde release refusée") + require(time.monotonic() < self.deadline, "release timeout") + self.released = True + self.connection.sendall(("RELEASE " + self.nonce).encode()) + message = self.connection.recv(4096) + require(message == ("EXEC_REQUESTED " + self.nonce).encode(), + "bootstrap EOF/exec intent absent") + + def close(self): + if self.connection is not None: + self.connection.close() + self.listener.close() + self.path.unlink(missing_ok=True) + + +def bootstrap(path, nonce, supervisor_pid, startticks, argv, proc=None, + execute=os.execve): + """Attendre avant prlimit/bwrap ; EOF, parent mort et nonce faux refusent.""" + proc = proc or ProcFS() + identity = proc.identity(supervisor_pid) + require(identity["uid"] == os.getuid() and identity["startticks"] == startticks, + "superviseur identité divergente") + descriptor = proc.pidfd(supervisor_pid) + try: + with socket.socket(socket.AF_UNIX, socket.SOCK_SEQPACKET) as channel: + channel.settimeout(4) + channel.connect(str(path)) + peer_pid, peer_uid, _ = struct.unpack("3i", channel.getsockopt( + socket.SOL_SOCKET, socket.SO_PEERCRED, struct.calcsize("3i"))) + require(peer_pid == supervisor_pid and peer_uid == os.getuid(), + "serveur étranger") + require(same_lifetime(identity, proc.identity(supervisor_pid)), + "PID superviseur recyclé") + channel.sendall(json.dumps({"nonce": nonce, "pid": os.getpid()}).encode()) + ready, _, _ = select.select([descriptor, channel], [], [], 4) + require(descriptor not in ready and channel in ready, "parent mort/timeout") + require(channel.recv(4096) == ("RELEASE " + nonce).encode(), + "EOF/nonce release invalide") + require(not select.select([descriptor], [], [], 0)[0], + "parent mort avant exec") + channel.sendall(("EXEC_REQUESTED " + nonce).encode()) + os.close(descriptor) + descriptor = None + # Do not rely on systemd-run's internal bus descriptor inheritance. + # close_fds at Popen protects the initial launch; close again at payload. + ceiling = resource.getrlimit(resource.RLIMIT_NOFILE)[1] + require(ceiling != resource.RLIM_INFINITY, "NOFILE hérité non fini") + os.closerange(3, ceiling) + os.chdir("/tmp") + execute(argv[0], argv, CLEAN_ENV) + raise Refusal("exec retourné sans remplacement") + finally: + if descriptor is not None: + os.close(descriptor) + + +def admit(spec, evidence, pid, proc, cgroups, manager, deadline): + """Figer l'identité complète avant RELEASE, puis revalider immédiatement.""" + group = evidence["app"] + "/" + spec["unit"] + observed = manager.show(spec["unit"], deadline) + scope_identity(spec, observed, group) + require(all(observed.get(key) == value for key, value in POLICY.items()), + "scope propriétés/budget divergents") + identity = proc.identity(pid) + require(identity["uid"] == evidence["uid"] and identity["cgroup"] == group, + "bootstrap PID/UID/membership divergent") + pidfd = proc.pidfd(pid) + try: + record = cgroups.snapshot(group) + require(record["uid"] == evidence["uid"] and record["pids_max"] == "256", + "pids.max non fini/incorrect ou propriétaire divergent") + require(record["procs"] == [pid] and record["pids_current"] == 1 + and not record["children"], + "scope occupé/pids.current incohérent") + require(same_lifetime(identity, proc.identity(pid)), "PID/startticks divergent") + require(proc.identity(evidence["launcher"]["pid"]) == evidence["launcher"], + "superviseur déplacé/identité divergente") + require(cgroups.mount() == evidence["mount"], "montage remplacé") + capacity(cgroups, evidence["app"]) + require(not select.select([pidfd], [], [], 0)[0], + "bootstrap mort avant admission") + current_manager = manager.show( + f"user@{evidence['uid']}.service", deadline, system=True) + require(all(current_manager.get(key) == evidence["manager"].get(key) + for key in ("MainPID", "InvocationID", "ControlGroup", "Delegate")), + "manager identité remplacée") + observed_manager = proc.identity(evidence["manager_process"]["pid"]) + require(observed_manager == evidence["manager_process"], + "manager PID/startticks divergent") + load = proc.uid_threads(evidence["uid"]) + require(proc.nproc() == evidence["nproc"] and all( + value == resource.RLIM_INFINITY or value >= load for value in proc.nproc()), + "NPROC/charge UID divergent avant release") + return {"properties": observed, "process": identity, "cgroup": group, + "kernel": record, "pidfd": pidfd} + except BaseException: + os.close(pidfd) + raise + + +def validate_owned(spec, receipt, cgroups, manager, deadline): + """Même invocation et inode ; un PID mort n'autorise jamais un nouveau PID.""" + observed = manager.show(spec["unit"], deadline) + scope_identity(spec, observed, receipt["cgroup"]) + require(observed["InvocationID"] == receipt["properties"]["InvocationID"], + "scope invocation remplacée") + kernel = cgroups.snapshot(receipt["cgroup"]) + require(all(kernel[key] == receipt["kernel"][key] + for key in ("inode", "device", "uid")), + "scope inode/propriétaire remplacé") + + +class EmptyWitness: + """Observer avant release et pendant stop, sans autorité de mutation. + + Un ENOENT/ENODEV n'est jamais converti en populated=0. Les lectures du + même inode doivent réellement constater events=0 et l'absence de tâches. + """ + + def __init__(self, cgroups, receipt): + self.cgroups = cgroups + self.receipt = receipt + self.empty_seen = False + self.errors = [] + self.finished = threading.Event() + self.thread = threading.Thread(target=self._observe, daemon=True) + self.thread.start() + + def sample(self): + try: + record = self.cgroups.snapshot(self.receipt["cgroup"]) + require(all(record[key] == self.receipt["kernel"][key] + for key in ("inode", "device", "uid")), + "witness inode divergent") + if self.cgroups.empty(self.receipt["cgroup"]): + # Recheck inode after recursive observation, before accepting proof. + after = self.cgroups.snapshot(self.receipt["cgroup"]) + require(all(after[key] == record[key] + for key in ("inode", "device", "uid")), + "witness inode remplacé pendant lecture") + self.empty_seen = True + except (FileNotFoundError, ProcessLookupError): + # Disappearance alone never establishes emptiness. + return + except BaseException as error: + self.errors.append(f"{type(error).__name__}: {error}"[:512]) + self.finished.set() + + def _observe(self): + while not self.finished.is_set(): + self.sample() + self.finished.wait(.002) + + def close(self): + self.finished.set() + self.thread.join(timeout=.1) + require(not self.thread.is_alive(), "witness arrêt non prouvé") + + +def cleanup(spec, receipt, process, proc, cgroups, manager, deadline, witness=None): + """Ne jamais laisser une exception de cleanup masquer son verdict fermé.""" + errors = [] + empty_seen = False + try: + if receipt is not None: + if witness is not None: + witness.sample() + empty_seen = witness.empty_seen + require(not witness.errors, "witness observation échouée") + observed = manager.show(spec["unit"], deadline) + exists = cgroups.exists(receipt["cgroup"]) + if observed.get("LoadState") == "not-found" and not exists: + require(empty_seen, "vide récursif non observé avant disparition") + else: + validate_owned(spec, receipt, cgroups, manager, deadline) + try: + current = proc.identity(receipt["process"]["pid"]) + except FileNotFoundError: + current = None + require(current is None or current == receipt["process"], + "PID/membership divergent avant stop") + empty_seen = empty_seen or cgroups.empty(receipt["cgroup"]) + manager.stop(spec["unit"], deadline) + while time.monotonic() < deadline: + if witness is not None: + witness.sample() + require(not witness.errors, "witness observation échouée") + empty_seen = empty_seen or witness.empty_seen + observed = manager.show(spec["unit"], deadline) + exists = cgroups.exists(receipt["cgroup"]) + if exists: + validate_owned(spec, receipt, cgroups, manager, deadline) + empty_seen = empty_seen or cgroups.empty(receipt["cgroup"]) + if observed.get("LoadState") == "not-found" and not exists: + require(empty_seen, + "vide récursif non observé avant disparition") + break + time.sleep(.02) + else: + raise Refusal("unit/cgroup disparition non prouvée") + else: + require(not cgroups.exists(evidence_group(spec)), + "scope créé sans ownership établi") + except BaseException as error: + errors.append(f"{type(error).__name__}: {error}"[:512]) + if witness is not None: + try: + witness.close() + except BaseException as error: + errors.append("witness close: " + str(error)[:128]) + if process is not None: + try: + if process.poll() is None: + process.kill() # Exact unreaped launcher child only. + process.wait(timeout=max(.001, deadline - time.monotonic())) + except BaseException as error: + errors.append(f"collection: {type(error).__name__}: {error}"[:512]) + for stream in (process.stdout, process.stderr): + try: + stream.close() + except BaseException as error: + errors.append("pipe close: " + str(error)[:128]) + if receipt is not None: + try: + os.close(receipt["pidfd"]) + except OSError as error: + errors.append("pidfd close: " + str(error)) + return {"proved": not errors, "empty_observed": empty_seen, "errors": errors} + + +def evidence_group(spec): + uid = os.getuid() + return f"/user.slice/user-{uid}.slice/user@{uid}.service/app.slice/" + spec["unit"] + + +def run_scope(lab, manifest, reserve, journal, proc=None, cgroups=None, + manager=None, barrier_factory=Barrier, collector=drain): + """Orchestration injectée : préflight, réserve, admission, release et récolte.""" + proc = proc or ProcFS() + cgroups = cgroups or CgroupFS() + manager = manager or SystemManager(os.getuid()) + spec = manifest["scope"] + started = time.monotonic() + process = barrier = receipt = witness = None + channels = [b"", b""] + result = {"cause": "failure", "phases": [], "runtime": "NOT_RUN", + "returncode": None, "collected": False} + + def phase(name, **detail): + result["phases"].append({"phase": name, **detail}) + journal(result) + + try: + evidence = preflight(proc, cgroups, manager, os.getuid(), os.getpid(), + started + 3) + group = evidence["app"] + "/" + spec["unit"] + require(manager.show(spec["unit"], started + 3).get("LoadState") == "not-found" + and not cgroups.exists(group), "scope déjà occupé ; aucun spawn") + # PRECHECK is not journaled until ATTEMPT: failed preflight stays read-only. + reserve(evidence) + phase("PRECHECK", nproc=evidence["nproc"], + uid_threads=evidence["uid_threads_diagnostic"]) + barrier = barrier_factory(lab / "gate.sock", spec["nonce"], os.getuid(), + os.getpid(), started + 5) + identity = evidence["launcher"] + bootstrap_argv = bootstrap_command(lab, spec, identity, manifest["argv"]) + process = manager.create(spec, bootstrap_argv) + phase("SCOPE_CREATED", status="creation_requested") + pid = barrier.accept() + receipt = admit(spec, evidence, pid, proc, cgroups, manager, started + 5) + public_receipt = {key: value for key, value in receipt.items() + if key != "pidfd"} + phase("ADMITTED", ownership=public_receipt) + witness = EmptyWitness(cgroups, receipt) + validate_owned(spec, receipt, cgroups, manager, started + 5) + require(proc.identity(pid) == receipt["process"], "PID drift avant release") + require(not select.select([receipt["pidfd"]], [], [], 0)[0], + "bootstrap mort avant release") + barrier.release() + phase("RELEASED", exec_intent=True) + if manifest.get("qualification_mode") == "parent-death": + # Only a separately sealed, separately approved benign LAB can reach this. + require(manifest["kind"] == "benign", "parent-death réservé au bénin") + phase("PARENT_DEATH_TEST", external_cleanup_required=True) + os._exit(71) + channels = collector(process, started + 8) + result["returncode"] = process.returncode + result["collected"] = process.returncode is not None + phase("COLLECTED", returncode=process.returncode) + if manifest["kind"] == "benign": + require(b"BENIGN_PAYLOAD_STARTED\n" in channels[0], + "marqueur payload bénin absent") + phase("PAYLOAD_STARTED", proof="benign stdout marker") + result["runtime"] = ("BENIGN" if manifest["kind"] == "benign" + else "EGL_RESULT_UNINTERPRETED") + result["cause"] = "exit" if process.returncode == 0 else "payload_failure" + except BaseException as error: + if isinstance(error, DrainFailure): + channels = error.channels + result["cause"] = ("interrupted" if isinstance(error, KeyboardInterrupt) + else "failure") + result["error"] = f"{type(error).__name__}: {error}"[:512] + finally: + if barrier is not None: + try: + barrier.close() + except BaseException as error: + result["barrier_cleanup_error"] = str(error)[:512] + if process is not None: + cleaned = cleanup(spec, receipt, process, proc, cgroups, manager, + started + 10, witness=witness) + result["cleanup"] = cleaned + if not cleaned["proved"] or "barrier_cleanup_error" in result: + result["cause"] = "CLEANUP_UNPROVEN" + try: + phase("CLEANED", proved=cleaned["proved"]) + except BaseException as error: + result["cause"] = "journal_failure" + result["error"] = str(error)[:512] + result["elapsed"] = time.monotonic() - started + return result, channels + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--bootstrap", nargs=argparse.REMAINDER, required=True) + args = parser.parse_args().bootstrap + require(len(args) >= 5, "bootstrap argv absent") + bootstrap(Path(args[0]), args[1], int(args[2]), int(args[3]), args[4:]) + + +if __name__ == "__main__": + try: + main() + except (Refusal, OSError, ValueError) as error: + print("Bootstrap refusé: " + str(error), file=sys.stderr) + sys.exit(70) diff --git a/tooling/probes/step20c-egl-fx/prepare.py b/tooling/probes/step20c-egl-fx/prepare.py index 11c125f..30fd02e 100644 --- a/tooling/probes/step20c-egl-fx/prepare.py +++ b/tooling/probes/step20c-egl-fx/prepare.py @@ -11,8 +11,9 @@ import shlex import shutil import subprocess import tarfile +import secrets -from run import ENV, HOST_ENV, command, digest, durable +from run import ENV, HOST_ENV, benign_command, command, digest, durable, gate PINS = { "scenefx": "37ccd723bef49e6891156ffafce8f549f01446cc", @@ -53,16 +54,29 @@ def archive_sources(bare, pin, destination): "verified_blobs": count} -def resolve_library(name): +def trusted_system_path(path): + """Refuser un chemin système dont un composant est modifiable hors root.""" + for candidate in {path, path.resolve()}: + for component in (candidate, *candidate.parents): + info = component.stat() + if info.st_uid != 0 or info.st_mode & 0o022: + raise ValueError(f"chemin système modifiable : {component}") + + +def resolve_library(name, runpath=()): if "/" in name: path = Path(name) if path.is_absolute() and path.is_relative_to("/usr") and path.exists(): return path raise ValueError(f"bibliothèque hors /usr : {name}") - path = Path("/usr/lib") / name - if not path.is_file(): - raise ValueError(f"dépendance ELF introuvable : {name}") - return path + # RUNPATH applies only to the declaring object's direct DT_NEEDED edges. + for directory in (*runpath, "/usr/lib"): + path = Path(directory) / name + if path.is_file(): + if runpath: + trusted_system_path(path) + return path + raise ValueError(f"dépendance ELF introuvable : {name}") def elf_closure(seeds): @@ -74,25 +88,53 @@ def elf_closure(seeds): path = pending.pop() if str(path) in result: continue - data = capture(["/usr/bin/readelf", "-dW", str(path)]).decode() - notes = capture(["/usr/bin/readelf", "-nW", str(path)]).decode() + data = capture(["/usr/bin/readelf", "-h", "-l", "-d", "-n", "-W", str(path)]).decode() needed = re.findall(r"\(NEEDED\).*\[([^]]+)\]", data) - build_id = re.search(r"Build ID: ([0-9a-f]+)", notes) - result[str(path)] = {"needed": needed, - "build_id": build_id.group(1) if build_id else None} + build_id = re.search(r"Build ID: ([0-9a-f]+)", data) + elf_class = re.search(r"Class:\s+(\S+)", data) + machine = re.search(r"Machine:\s+([^\n]+)", data) + interpreter = re.findall(r"Requesting program interpreter: ([^\]]+)", data) + runpath = re.findall(r"\(RUNPATH\).*\[([^]]*)\]", data) + if runpath: + # Exact distro systemd exception; no expansion or generic fallback. + if (runpath != ["/usr/lib/systemd"] or + (path not in (Path("/usr/bin/systemctl"), Path("/usr/bin/systemd-run")) + and path.parent != Path("/usr/lib/systemd"))): + raise ValueError(f"RUNPATH ELF refusé : {path}") + trusted_system_path(path) + trusted_system_path(Path(runpath[0])) + if (not elf_class or elf_class.group(1) != "ELF64" or not machine + or machine.group(1).strip() != "Advanced Micro Devices X86-64" + or not build_id or re.search(r"\(RPATH\)", data)): + raise ValueError(f"ABI/Build ID/RPATH ELF refusé : {path}") + if interpreter and interpreter not in ( + ["/lib64/ld-linux-x86-64.so.2"], ["/usr/lib/ld-linux-x86-64.so.2"]): + raise ValueError(f"interpréteur ELF refusé : {path}") + result[str(path)] = { + "needed": needed, "build_id": build_id.group(1), + "class": elf_class.group(1), "machine": machine.group(1).strip(), + "interpreter": interpreter, + "runpath": runpath, + } canonical = path.resolve() if canonical not in seen: seen.add(canonical) - pending.extend(resolve_library(name) for name in needed) + pending.extend(resolve_library(name, runpath) for name in needed) return result def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--lab", type=Path, required=True) - parser.add_argument("--source-store", type=Path, required=True) + parser.add_argument("--source-store", type=Path) + parser.add_argument("--benign", action="store_true") + parser.add_argument("--qualification-mode", choices=("nominal", "parent-death"), default="nominal") args = parser.parse_args() lab = args.lab + if not args.benign and args.source_store is None: + raise ValueError("--source-store requis pour la préparation EGL") + if not args.benign and args.qualification_mode != "nominal": + raise ValueError("parent-death réservé au bénin") if not lab.is_absolute() or lab.resolve() != lab: raise ValueError("LAB absolu sans symlink requis") if lab.exists(): @@ -104,8 +146,27 @@ def main(): lab.mkdir(mode=0o700) for name in ("fixture", "sources", "generated", "bin", "evidence"): (lab / name).mkdir(exist_ok=True) - for name in ("probe.c", "run.py", "prepare.py", "test_probe.py", "test_probe_mocks.c", "README.md"): + for name in ("probe.c", "run.py", "cgroup_gate.py", "prepare.py", "test_probe.py", + "test_cgroup_admission.py", "test_probe_mocks.c", "README.md"): shutil.copyfile(FIXTURE / name, lab / "fixture" / name) + if args.benign: + files = {path for path in (lab / "fixture").iterdir() if path.is_file()} + tools = [Path("/usr/bin") / name for name in + ("python3", "systemd-run", "systemctl", "env", "readelf")] + closure = elf_closure(tools) + files.update(tools) + files.update(Path(name) for name in closure) + manifest = { + "schema": 2, "kind": "benign", "qualification_mode": args.qualification_mode, + "lab": str(lab), "runtime": "NOT_RUN", "budget": 1, + "argv": benign_command(args.qualification_mode), "environment": HOST_ENV, + "supervisor_environment": HOST_ENV, "scope": gate.policy(secrets.token_hex(16)), + "elf_closure": closure, + "files": {str(path): {"sha256": digest(path), "resolved": str(path.resolve())} + for path in sorted(files)}, + } + publish(lab, manifest) + return provenance = {} for name, pin in PINS.items(): provenance[name] = archive_sources(args.source_store / (name + ".git"), pin, @@ -149,7 +210,8 @@ def main(): # Mesa uses dlopen as well as DT_NEEDED. Seal explicit software DRI, # GBM plugin and gallium provider, then their entire DT_NEEDED closure. seeds = [lab / "bin/probe", Path("/usr/bin/bwrap"), Path("/usr/bin/prlimit"), - Path("/usr/bin/python3"), Path("/usr/bin/env"), Path("/usr/lib/ld-linux-x86-64.so.2"), + Path("/usr/bin/python3"), Path("/usr/bin/systemd-run"), + Path("/usr/bin/systemctl"), Path("/usr/bin/env"), Path("/usr/lib/ld-linux-x86-64.so.2"), vendor_library, Path("/usr/lib/dri/swrast_dri.so")] seeds.extend(Path("/usr/lib").glob("libgallium-*.so")) seeds.append(Path("/usr/lib/gbm/dri_gbm.so")) @@ -176,7 +238,9 @@ def main(): identities = {str(path): {"sha256": digest(path), "resolved": str(path.resolve())} for path in sorted(files)} manifest = { - "schema": 1, "lab": str(lab), "runtime": "NOT_RUN", "budget": 1, + "schema": 2, "kind": "egl", "qualification_mode": "nominal", + "scope": gate.policy(secrets.token_hex(16)), + "lab": str(lab), "runtime": "NOT_RUN", "budget": 1, "argv": command(lab), "environment": ENV, "supervisor_environment": HOST_ENV, "pins": provenance, "files": identities, "elf_closure": closure, "symbols": symbols, "compile_commands": commands, @@ -186,6 +250,12 @@ def main(): "dlopen_limit": "Explicit Mesa EGL, swrast, GBM and gallium providers plus DT_NEEDED " "are sealed; conditional runtime dlopen discovery remains NOT_RUN.", } + publish(lab, manifest) + + +def publish(lab, manifest): + """Sceller les octets préparés et imprimer seulement une commande passive.""" + manifest["scope_argv"] = gate.sealed_scope_argv(lab, manifest) durable(lab / "manifest.json", json.dumps(manifest, indent=2, sort_keys=True).encode() + b"\n") seal = digest(lab / "manifest.json") print(json.dumps({"runtime": "NOT_RUN", "manifest_sha256": seal, "plan_command": diff --git a/tooling/probes/step20c-egl-fx/run.py b/tooling/probes/step20c-egl-fx/run.py index 1808ffa..27f02c1 100644 --- a/tooling/probes/step20c-egl-fx/run.py +++ b/tooling/probes/step20c-egl-fx/run.py @@ -11,6 +11,26 @@ import stat import subprocess import sys import time +import types + +class _SealedHelper: + """Chargement différé : le digest du helper précède son import dans main.""" + + module = None + + def __getattr__(self, name): + if self.module is None: + path = Path(__file__).resolve().with_name("cgroup_gate.py") + # SourceFileLoader can write __pycache__ even for a passive plan. + # Compile the sealed sibling bytes directly, without a loader cache. + module = types.ModuleType("mosaic_cgroup_gate") + module.__file__ = str(path) + exec(compile(path.read_bytes(), str(path), "exec"), module.__dict__) + self.module = module + return getattr(self.module, name) + + +gate = _SealedHelper() LIMIT = 1024 * 1024 ENV = { @@ -35,7 +55,7 @@ def command(lab): """Liste fermée : aucun mount, argument ou environnement fourni par l'appelant.""" argv = [ "/usr/bin/prlimit", "--as=1073741824", "--cpu=5", "--nofile=64", - "--fsize=1048576", "--core=0", "--nproc=256", "--", + "--fsize=1048576", "--core=0", "--", "/usr/bin/bwrap", "--unshare-user", "--unshare-all", "--disable-userns", "--new-session", "--die-with-parent", "--clearenv", "--ro-bind", "/usr", "/usr", "--symlink", "usr/bin", "/bin", "--symlink", "usr/lib", "/lib", @@ -48,6 +68,23 @@ def command(lab): return argv + ["--", "/probe"] +def benign_command(mode): + """Deux charges fixes, inoffensives, sans Bubblewrap ni bibliothèque EGL.""" + if mode not in ("nominal", "parent-death"): + raise Refusal("qualification mode invalide") + delay = "0.2" if mode == "nominal" else "30" + if mode == "parent-death": + # Le superviseur peut fermer le pipe avant l'écriture ; le délai doit + # quand même laisser managerRuntime12 observer la mort du parent. + script = ("import os,time\n" + "try:\n os.write(1,b'BENIGN_PAYLOAD_STARTED\\n')\n" + "finally:\n time.sleep(30)") + else: + script = ("import os,time; os.write(1,b'BENIGN_PAYLOAD_STARTED\\n'); " + "time.sleep(" + delay + ")") + return ["/usr/bin/python3", "-I", "-S", "-B", "-c", script] + + def verify(lab, expected): """Lecture seule, y compris pour le plan ; le digest doit venir de la revue.""" if dict(os.environ) != HOST_ENV: @@ -61,14 +98,33 @@ def verify(lab, expected): if digest(manifest_path) != expected: raise Refusal("digest du manifeste différent de l'approbation") manifest = json.loads(manifest_path.read_text()) - if manifest["schema"] != 1 or manifest["lab"] != str(lab): + if manifest["schema"] != 2 or manifest["lab"] != str(lab): raise Refusal("identité du manifeste") - if manifest["argv"] != command(lab) or manifest["environment"] != ENV: + kind = manifest["kind"] + if kind not in ("benign", "egl"): + raise Refusal("type de charge invalide") + expected_argv = (benign_command(manifest["qualification_mode"]) + if kind == "benign" else command(lab)) + expected_env = HOST_ENV if kind == "benign" else ENV + helper = lab / "fixture/cgroup_gate.py" + helper_identity = manifest["files"].get(str(helper)) + if (not helper_identity or digest(helper) != helper_identity["sha256"] + or str(helper.resolve()) != helper_identity["resolved"]): + raise Refusal("identité helper altérée avant import") + if manifest["scope_argv"] != gate.sealed_scope_argv(lab, manifest): + raise Refusal("argv du scope altéré") + if manifest["scope"] != gate.policy(manifest["scope"]["nonce"]): + raise Refusal("politique scope altérée") + if manifest["argv"] != expected_argv or manifest["environment"] != expected_env: raise Refusal("montages, devices ou environnement altérés") if manifest["supervisor_environment"] != HOST_ENV: raise Refusal("environnement du superviseur altéré") - required = {str(lab / "bin/probe"), str(lab / "fixture/run.py"), - "/usr/bin/bwrap", "/usr/bin/prlimit"} + required = {str(lab / "fixture/run.py"), + str(lab / "fixture/cgroup_gate.py"), "/usr/bin/python3", + "/usr/bin/systemd-run", "/usr/bin/systemctl", "/usr/bin/env", + "/usr/bin/readelf"} + if kind == "egl": + required.update({str(lab / "bin/probe"), "/usr/bin/bwrap", "/usr/bin/prlimit"}) if not required.issubset(manifest["files"]): raise Refusal("identités requises absentes") if Path(__file__).resolve() != lab / "fixture/run.py": @@ -79,6 +135,8 @@ def verify(lab, expected): raise Refusal("dépendance hors LAB neuf ou /usr") if digest(path) != identity["sha256"]: raise Refusal(f"dérive identité : {path}") + if not (path.resolve().is_relative_to(lab) or path.resolve().is_relative_to("/usr")): + raise Refusal(f"dépendance résolue hors LAB ou /usr : {path}") if str(path.resolve()) != identity["resolved"]: raise Refusal(f"dérive symlink : {path}") return manifest @@ -111,7 +169,8 @@ def supervise(argv, payload_seconds=8.0, total_seconds=10.0): bwrap conserve le PID namespace jusqu'à la fin du payload ; sa disparition termine les descendants de ce namespace. Pas de kill d'un PID déjà récolté. - Les limites AS/CPU sont par processus ; NPROC porte sur l'UID, pas un cgroup. + Les limites AS/CPU sont par processus ; NPROC hérité porte sur l'UID, pas un cgroup. + Ce collecteur historique est testé séparément ; main utilise gate.run_scope. """ started = time.monotonic() channels = [bytearray(), bytearray()] @@ -212,29 +271,42 @@ def main(): parser.add_argument("--lab", required=True, type=Path) parser.add_argument("--manifest-sha256", required=True) parser.add_argument("--approve-native-probe", action="store_true") + parser.add_argument("--approve-benign-gate", action="store_true") args = parser.parse_args() try: manifest = verify(args.lab, args.manifest_sha256) if (args.lab / "ATTEMPT").exists(): raise Refusal("budget déjà consommé ; aucun retry") - if not args.approve_native_probe: + approved = args.approve_native_probe or args.approve_benign_gate + if not approved: print(json.dumps({"runtime": "NOT_RUN", "budget": 1, - "argv": manifest["argv"], "environment": ENV}, indent=2)) + "argv": manifest["argv"], "scope": manifest["scope"], + "environment": manifest["environment"]}, indent=2)) return 0 - durable(args.lab / "ATTEMPT", json.dumps({ - "manifest_sha256": args.manifest_sha256, "budget_consumed": 1, - }).encode() + b"\n", exclusive=True) - try: - result, channels = supervise(manifest["argv"]) - except OSError as error: - result = {"cause": "spawn_failure", "error": str(error), "collected": True} - channels = [b"", b""] + if (args.approve_native_probe == args.approve_benign_gate or + args.approve_benign_gate != (manifest["kind"] == "benign")): + raise Refusal("drapeau approbation incompatible avec le manifeste") + + def reserve(evidence): + durable(args.lab / "ATTEMPT", json.dumps({ + "manifest_sha256": args.manifest_sha256, "budget_consumed": 1, + "scope": manifest["scope"], "preflight": evidence, + }).encode() + b"\n", exclusive=True) + + def journal(result): + durable(args.lab / "result.json", json.dumps(result, sort_keys=True).encode() + b"\n") + + result, channels = gate.run_scope(args.lab, manifest, reserve, journal) + if not (args.lab / "ATTEMPT").exists(): + # Failed read-only preflight does not consume or mutate the LAB. + print(json.dumps(result)) + return 20 for name, data in zip(("stdout.raw", "stderr.raw"), channels): durable(args.lab / name, data) durable(args.lab / "result.json", json.dumps(result, sort_keys=True).encode() + b"\n") print(json.dumps(result)) return 0 if result.get("cause") == "exit" and result.get("returncode") == 0 else 20 - except (Refusal, OSError, ValueError, KeyError, TypeError) as error: + except (RuntimeError, Refusal, OSError, ValueError, KeyError, TypeError) as error: print(f"Refus : {error}", file=sys.stderr) return 20 diff --git a/tooling/probes/step20c-egl-fx/test_cgroup_admission.py b/tooling/probes/step20c-egl-fx/test_cgroup_admission.py new file mode 100644 index 0000000..baf41e7 --- /dev/null +++ b/tooling/probes/step20c-egl-fx/test_cgroup_admission.py @@ -0,0 +1,957 @@ +"""Tests des fonctions de production ; aucune création native n'est permise.""" +import contextlib +import copy +import io +import json +import os +from pathlib import Path +import resource +import socket +import subprocess +import sys +import tempfile +import threading +import time +import unittest +from unittest.mock import patch + +import run + +INERT_POPEN = subprocess.Popen +gate = run.gate +UID = os.getuid() +PID = os.getpid() +BASE = f"/user.slice/user-{UID}.slice/user@{UID}.service" +APP = BASE + "/app.slice" +SPEC = gate.policy("a" * 32) +gate = run.gate.module # Patch the production module globals, not the lazy proxy. +GROUP = APP + "/" + SPEC["unit"] + + +def native_spawn_forbidden(*args, **kwargs): + raise AssertionError("native_spawn_forbidden") + + +_native_sentinel = None + + +def setUpModule(): + global _native_sentinel + _native_sentinel = patch.object( + subprocess, "Popen", side_effect=native_spawn_forbidden) + _native_sentinel.start() + + +def tearDownModule(): + _native_sentinel.stop() + + +class FakeProc: + def __init__(self): + self.identities = { + PID: {"pid": PID, "uid": UID, "startticks": 111, + "cgroup": APP + "/launcher.scope"}, + 42: {"pid": 42, "uid": UID, "startticks": 222, + "cgroup": BASE + "/init.scope"}, + 43: {"pid": 43, "uid": UID, "startticks": 333, "cgroup": GROUP}, + } + self.limits = (4096, 4096) + self.load = 1126 + self.writers = [] + + def identity(self, pid): + return dict(self.identities[pid]) + + def uid_threads(self, uid): + return self.load + + def nproc(self): + return self.limits + + def pidfd(self, pid): + reader, writer = os.pipe() + self.writers.append(writer) + return reader + + def close(self): + for writer in self.writers: + os.close(writer) + + +class FakeCgroups: + def __init__(self): + self.mount_id = (1, "0:30") + self.controllers = (["pids"], ["pids"], True) + self.present = False + self.is_empty = True + self.record = {"inode": 5, "device": 30, "uid": UID, + "pids_max": "256", "pids_current": 1, "procs": [43], + "children": []} + self.ancestor = {**self.record, "pids_max": "4096", "pids_current": 3} + + def mount(self): + return self.mount_id + + def snapshot(self, group): + return dict(self.record if group == GROUP else self.ancestor) + + def available(self, group): + return self.controllers + + def exists(self, group): + return self.present + + def empty(self, group): + return self.is_empty + + +class FakeProcess: + def __init__(self): + self.returncode = None + self.stdout = io.BytesIO() + self.stderr = io.BytesIO() + self.killed = False + self.wait_error = None + + def poll(self): + return self.returncode + + def kill(self): + self.killed = True + self.returncode = -9 + + def wait(self, timeout): + if self.wait_error: + raise self.wait_error + self.returncode = 0 if self.returncode is None else self.returncode + return self.returncode + + +class FakeManager: + def __init__(self, cgroups): + self.cgroups = cgroups + self.created = 0 + self.stopped = [] + self.owner = {"Id": f"user@{UID}.service", "Delegate": "yes", + "ActiveState": "active", + "ControlGroup": BASE, "InvocationID": "b" * 32, "MainPID": "42"} + self.app = {"Id": "app.slice", "ActiveState": "active", "ControlGroup": APP} + self.scope = {"Id": SPEC["unit"], "Description": SPEC["description"], + "InvocationID": "c" * 32, "ControlGroup": GROUP, **gate.POLICY} + self.process = FakeProcess() + self.create_error = None + self.stop_error = None + + def show(self, unit, deadline, system=False): + if unit == f"user@{UID}.service": + return dict(self.owner) + if unit == "app.slice": + return dict(self.app) + return dict(self.scope) if self.cgroups.present else {"LoadState": "not-found"} + + def create(self, spec, argv): + self.created += 1 + if self.create_error: + raise self.create_error + self.cgroups.present = True + return self.process + + def stop(self, unit, deadline): + if self.stop_error: + raise self.stop_error + self.stopped.append(unit) + self.cgroups.present = False + + +class FakeBarrier: + def __init__(self, *args): + self.released = 0 + self.closed = False + self.error = None + + def accept(self): + if self.error: + raise self.error + return 43 + + def release(self): + self.released += 1 + + def close(self): + self.closed = True + + +class AdmissionTests(unittest.TestCase): + def setUp(self): + self.sentinel = patch.object( + subprocess, "Popen", side_effect=native_spawn_forbidden) + self.sentinel.start() + self.addCleanup(self.sentinel.stop) + self.proc = FakeProc() + self.addCleanup(self.proc.close) + self.cgroups = FakeCgroups() + self.manager = FakeManager(self.cgroups) + self.barrier = FakeBarrier() + self.temporary = tempfile.TemporaryDirectory() + self.addCleanup(self.temporary.cleanup) + self.lab = Path(self.temporary.name) + self.manifest = {"scope": SPEC, "kind": "benign", + "qualification_mode": "nominal", + "argv": run.benign_command("nominal")} + self.reservations = [] + self.journal = [] + + def preflight(self): + return gate.preflight(self.proc, self.cgroups, self.manager, + UID, PID, time.monotonic() + 1) + + def admitted(self): + evidence = self.preflight() + self.cgroups.present = True + return gate.admit(SPEC, evidence, 43, self.proc, self.cgroups, self.manager, + time.monotonic() + 1) + + def execute(self, collector=None, reserve=None, journal=None): + def collect(process, deadline): + process.returncode = 0 + return [b"BENIGN_PAYLOAD_STARTED\n", b""] + return gate.run_scope( + self.lab, self.manifest, reserve or self.reservations.append, + journal or (lambda result: self.journal.append(copy.deepcopy(result))), + self.proc, self.cgroups, self.manager, + barrier_factory=lambda *args: self.barrier, collector=collector or collect) + + def test_success_uses_production_orchestrator_and_single_release(self): + result, output = self.execute() + self.assertEqual(result["cause"], "exit") + self.assertEqual(self.barrier.released, 1) + self.assertEqual(len(self.reservations), 1) + self.assertTrue(self.barrier.closed) + self.assertEqual(self.manager.stopped, [SPEC["unit"]]) + self.assertEqual([event["phase"] for event in result["phases"]], + ["PRECHECK", "SCOPE_CREATED", "ADMITTED", "RELEASED", + "COLLECTED", + "PAYLOAD_STARTED", "CLEANED"]) + self.assertEqual(self.proc.limits, (4096, 4096)) + + def test_session_scope_refused_before_reservation(self): + self.proc.identities[PID]["cgroup"] = ( + f"/user.slice/user-{UID}.slice/session-6.scope") + result, _ = self.execute() + self.assertIn("session scope", result["error"]) + self.assertEqual(self.reservations, []) + self.assertEqual(self.manager.created, 0) + + def test_nproc_256_under_uid1126_refused_and_never_raised(self): + self.proc.limits = (256, 256) + result, _ = self.execute() + self.assertIn("RLIMIT_NPROC", result["error"]) + self.assertEqual(self.proc.limits, (256, 256)) + self.assertEqual(self.manager.created, 0) + + def test_infinite_inherited_nproc_is_preserved(self): + self.proc.limits = (resource.RLIM_INFINITY, resource.RLIM_INFINITY) + self.assertEqual(self.preflight()["nproc"], self.proc.limits) + self.assertFalse(any(arg.startswith("--nproc") + for arg in run.command(self.lab))) + + def test_manager_delegation_identity_and_app_refusals(self): + cases = (("owner", "Delegate", "no"), ("owner", "Id", "foreign"), + ("owner", "InvocationID", ""), ("owner", "ControlGroup", "/"), + ("app", "ActiveState", "inactive"), ("app", "ControlGroup", "/")) + for target, key, bad in cases: + with self.subTest(key=key, bad=bad): + original = getattr(self.manager, target)[key] + getattr(self.manager, target)[key] = bad + with self.assertRaises(gate.Refusal): + self.preflight() + getattr(self.manager, target)[key] = original + + def test_controllers_and_writable_ancestor(self): + for controllers in (([], ["pids"], True), (["pids"], [], True), + (["pids"], ["pids"], False)): + self.cgroups.controllers = controllers + with self.assertRaises(gate.Refusal): + self.preflight() + + def test_ancestor_headroom_and_inconsistent_current(self): + for maximum, current in (("1200", 1126), ("4096", -1), ("broken", 0)): + self.cgroups.ancestor.update(pids_max=maximum, pids_current=current) + with self.assertRaises(gate.Refusal): + self.preflight() + + def test_scope_collision_no_spawn(self): + self.cgroups.present = True + result, _ = self.execute() + self.assertIn("occupé", result["error"]) + self.assertEqual(self.manager.created, 0) + self.assertEqual(self.reservations, []) + + def test_uid_divergence_before_release(self): + self.proc.identities[43]["uid"] = UID + 1 + result, _ = self.execute() + self.assertEqual(self.barrier.released, 0) + self.assertEqual(result["cause"], "CLEANUP_UNPROVEN") + self.assertEqual(self.manager.stopped, []) + + def test_kernel_budget_membership_and_shape_refusals(self): + for key, bad in (("pids_max", "max"), ("pids_max", "257"), ("pids_current", 0), + ("pids_current", 2), ("procs", [43, 44]), + ("children", ["foreign"]), + ("uid", UID + 1)): + with self.subTest(key=key, bad=bad): + original = self.cgroups.record[key] + self.cgroups.record[key] = bad + with self.assertRaises(gate.Refusal): + self.admitted() + self.cgroups.record[key] = original + + def test_scope_manager_properties_are_exact(self): + for key, bad in (("Id", "foreign.scope"), ("Description", "other"), + ("InvocationID", ""), ("ControlGroup", "/wrong"), + ("TasksMax", "infinity"), ("Delegate", "no"), + ("RuntimeMaxUSec", "infinity"), ("KillMode", "process")): + with self.subTest(key=key): + original = self.manager.scope[key] + self.manager.scope[key] = bad + with self.assertRaises(gate.Refusal): + self.admitted() + self.manager.scope[key] = original + + def test_startticks_change_during_admission(self): + original = self.proc.identity + calls = 0 + def changing(pid): + nonlocal calls + value = original(pid) + if pid == 43: + calls += 1 + if calls > 1: + value["startticks"] += 1 + return value + self.proc.identity = changing + with self.assertRaisesRegex(gate.Refusal, "startticks"): + self.admitted() + + def test_dead_pidfd_refused(self): + def dead(pid): + reader, writer = os.pipe() + os.close(writer) + return reader + self.proc.pidfd = dead + with self.assertRaisesRegex(gate.Refusal, "mort"): + self.admitted() + + def test_identity_drift_never_stops_foreign_scope(self): + for key, bad in (("Id", "foreign.scope"), ("Description", "other"), + ("InvocationID", "d" * 32), ("ControlGroup", "/foreign")): + with self.subTest(key=key): + self.manager.scope = {"Id": SPEC["unit"], + "Description": SPEC["description"], + "InvocationID": "c" * 32, "ControlGroup": GROUP, + **gate.POLICY} + receipt = self.admitted() + self.manager.scope[key] = bad + cleaned = gate.cleanup(SPEC, receipt, None, self.proc, self.cgroups, + self.manager, time.monotonic() + .2) + self.assertFalse(cleaned["proved"]) + self.assertEqual(self.manager.stopped, []) + + def test_inode_drift_and_recycled_pid_never_stop(self): + receipt = self.admitted() + self.cgroups.record["inode"] += 1 + self.assertFalse(gate.cleanup(SPEC, receipt, None, self.proc, self.cgroups, + self.manager, time.monotonic() + .2)["proved"]) + self.assertEqual(self.manager.stopped, []) + self.cgroups.record["inode"] -= 1 + receipt = self.admitted() + self.proc.identities[43]["startticks"] += 1 + self.assertFalse(gate.cleanup(SPEC, receipt, None, self.proc, self.cgroups, + self.manager, time.monotonic() + .2)["proved"]) + self.assertEqual(self.manager.stopped, []) + + def test_eof_and_timeout_before_release_never_exec(self): + for error in (EOFError("child EOF"), TimeoutError("barrier timeout")): + self.barrier.error = error + result, _ = self.execute() + self.assertNotEqual(result["cause"], "exit") + self.assertEqual(self.barrier.released, 0) + + def test_failures_in_preflight_reservation_admission_and_drain(self): + for error in (OSError("read failure"), KeyboardInterrupt()): + with patch.object(self.proc, "uid_threads", side_effect=error): + result, _ = self.execute() + self.assertNotEqual(result["cause"], "exit") + self.assertEqual(self.manager.created, 0) + def fail(*args): + raise OSError("mock failure") + result, _ = self.execute(reserve=fail) + self.assertEqual(self.manager.created, 0) + self.assertNotEqual(result["cause"], "exit") + result, _ = self.execute(collector=fail) + self.assertNotEqual(result["cause"], "exit") + self.assertTrue(self.manager.process.stdout.closed) + self.assertTrue(self.manager.process.stderr.closed) + + def test_spawn_failure_consumes_once_no_cleanup_exception(self): + self.manager.create_error = OSError("Popen failure") + result, _ = self.execute() + self.assertEqual(len(self.reservations), 1) + self.assertIn("Popen failure", result["error"]) + self.assertEqual(self.manager.stopped, []) + + def test_cleanup_stop_collection_and_events_failures_never_pass(self): + for mode in ("stop", "collection", "populated"): + with self.subTest(mode=mode): + self.manager = FakeManager(self.cgroups) + self.cgroups.present = False + self.cgroups.is_empty = mode != "populated" + if mode == "stop": + self.manager.stop_error = OSError("stop failed") + if mode == "collection": + self.manager.process.wait_error = subprocess.TimeoutExpired( + "mock", .1) + result, _ = self.execute() + self.assertEqual(result["cause"], "CLEANUP_UNPROVEN") + + def test_journal_failure_does_not_abandon_child(self): + calls = 0 + def journal(result): + nonlocal calls + calls += 1 + if calls >= 4: + raise OSError("journal full") + result, _ = self.execute(journal=journal) + self.assertNotEqual(result["cause"], "exit") + self.assertTrue(self.manager.process.stdout.closed) + self.assertTrue(self.barrier.closed) + + def test_no_egl_measurement_when_only_exec_requested(self): + self.manifest["kind"] = "egl" + result, _ = self.execute() + self.assertEqual(result["runtime"], "EGL_RESULT_UNINTERPRETED") + self.assertNotIn("PAYLOAD_STARTED", + [event["phase"] for event in result["phases"]]) + + def test_parent_death_is_separate_sealed_case(self): + self.manifest["qualification_mode"] = "parent-death" + with patch.object(gate.os, "_exit", side_effect=SystemExit(71)) as exit_process: + result, _ = self.execute() + exit_process.assert_called_once_with(71) + self.assertNotEqual(result["cause"], "exit") + self.assertIn("PARENT_DEATH_TEST", + [event["phase"] for event in result["phases"]]) + + def test_empty_observed_before_unit_disappears_allows_collection(self): + def collect(process, deadline): + self.cgroups.is_empty = True + self.cgroups.record.update(pids_current=0, procs=[]) + time.sleep(.01) # Give the production watcher a real observation. + self.cgroups.present = False + process.returncode = 0 + return [b"BENIGN_PAYLOAD_STARTED\n", b""] + self.cgroups.is_empty = False + result, _ = self.execute(collector=collect) + self.assertEqual(result["cause"], "exit") + self.assertTrue(result["cleanup"]["empty_observed"]) + self.assertEqual(self.manager.stopped, []) + + def test_disappeared_without_empty_proof_never_passes(self): + def collect(process, deadline): + self.cgroups.present = False + process.returncode = 0 + return [b"BENIGN_PAYLOAD_STARTED\n", b""] + self.cgroups.is_empty = False + result, _ = self.execute(collector=collect) + self.assertEqual(result["cause"], "CLEANUP_UNPROVEN") + self.assertEqual(self.manager.stopped, []) + + def test_drain_failure_keeps_captured_prefix_in_orchestrator(self): + def collect(process, deadline): + raise gate.DrainFailure("overflow", [b"prefix", b"diagnostic"]) + result, channels = self.execute(collector=collect) + self.assertNotEqual(result["cause"], "exit") + self.assertEqual(channels, [b"prefix", b"diagnostic"]) + + + def test_approved_main_uses_production_gate_and_durable_single_budget(self): + def execute(lab, manifest, reserve, journal): + def collect(process, deadline): + process.returncode = 0 + return [b"BENIGN_PAYLOAD_STARTED\n", b""] + return gate.run_scope( + lab, manifest, reserve, journal, self.proc, self.cgroups, self.manager, + barrier_factory=lambda *args: self.barrier, collector=collect) + + argv = ["run.py", "--lab", str(self.lab), "--manifest-sha256", "reviewed", + "--approve-benign-gate"] + with ( + patch.object(sys, "argv", argv), + patch.object(run, "verify", return_value=self.manifest), + patch.object(run.gate, "run_scope", side_effect=execute), + contextlib.redirect_stdout(io.StringIO()), + contextlib.redirect_stderr(io.StringIO()), + ): + self.assertEqual(run.main(), 0) + self.assertEqual(run.main(), 20) + self.assertEqual(self.manager.created, 1) + receipt = json.loads((self.lab / "result.json").read_text()) + self.assertEqual(receipt["returncode"], 0) + self.assertTrue(receipt["cleanup"]["proved"]) + self.assertTrue((self.lab / "ATTEMPT").exists()) + + def test_missing_benign_payload_marker_cannot_pass(self): + def collect(process, deadline): + process.returncode = 0 + return [b"", b""] + result, _ = self.execute(collector=collect) + self.assertNotEqual(result["cause"], "exit") + self.assertIn("marqueur", result["error"]) + + + +class FilesystemTests(unittest.TestCase): + def setUp(self): + self.temp = tempfile.TemporaryDirectory() + self.addCleanup(self.temp.cleanup) + self.root = Path(self.temp.name) + self.fs = gate.CgroupFS(self.root) + self.group = self.root / "scope" + self.group.mkdir() + contents = {"cgroup.type": "domain", "pids.max": "256", "pids.current": "1", + "cgroup.procs": "43", "cgroup.events": "populated 0\n", + "cgroup.threads": ""} + for name, value in contents.items(): + (self.group / name).write_text(value) + + def test_threaded_and_symlink_rejected_by_production_fs(self): + (self.group / "cgroup.type").write_text("threaded") + with self.assertRaisesRegex(gate.Refusal, "domain"): + self.fs.snapshot("/scope") + + + (self.root / "link").symlink_to(self.group) + with self.assertRaisesRegex(gate.Refusal, "symlink"): + self.fs.snapshot("/link") + + def test_cgroup_v1_mount_and_nonroot_mount_rejected(self): + for text in (f"1 0 0:30 / {self.root} rw - cgroup cgroup rw\n", + f"1 0 0:30 /subtree {self.root} rw - cgroup2 cgroup rw\n"): + with patch.object(Path, "read_text", return_value=text): + with self.assertRaises(gate.Refusal): + self.fs.mount() + + def test_recursive_populated_threads_and_symlink(self): + (self.group / "cgroup.procs").write_text("") + (self.group / "pids.current").write_text("0") + self.assertTrue(self.fs.empty("/scope")) + (self.group / "cgroup.threads").write_text("44") + self.assertFalse(self.fs.empty("/scope")) + (self.group / "cgroup.threads").write_text("") + (self.group / "foreign").symlink_to(self.root) + with self.assertRaises(gate.Refusal): + self.fs.empty("/scope") + + def test_proc_v1_and_uid_drift_refused(self): + directory = self.root / "43" + directory.mkdir() + (directory / "stat").write_text("43 (name) " + " ".join(["1"] * 20)) + (directory / "status").write_text( + f"Uid:\t{UID} {UID} {UID} {UID}\nThreads:\t1\n") + (directory / "cgroup").write_text("1:name:/legacy\n") + proc = gate.ProcFS(self.root) + with self.assertRaisesRegex(gate.Refusal, "v1"): + proc.identity(43) + (directory / "cgroup").write_text("0::/scope\n") + (directory / "status").write_text( + f"Uid:\t{UID} {UID + 1} {UID} {UID}\nThreads:\t1\n") + with self.assertRaisesRegex(gate.Refusal, "UID"): + proc.identity(43) + + def test_proc_uid_thread_diagnostic_rejects_drift_and_disappearance(self): + directory = self.root / "43" + directory.mkdir() + (directory / "stat").write_text("43 (name) " + " ".join(["1"] * 20)) + status = f"Uid:\t{UID} {UID} {UID} {UID}\nThreads:\t1\n" + (directory / "status").write_text(status) + (directory / "cgroup").write_text("0::/scope\n") + (directory / "task/43").mkdir(parents=True) + proc = gate.ProcFS(self.root) + self.assertEqual(proc.uid_threads(UID), 1) + (directory / "task/44").mkdir() + with self.assertRaisesRegex(gate.Refusal, "threads incohérent"): + proc.uid_threads(UID) + (directory / "task/44").rmdir() + original = Path.read_text + reads = 0 + + def changed(path, *args, **kwargs): + nonlocal reads + if path == directory / "status": + reads += 1 + if reads > 2: + return status.replace("Threads:\t1", "Threads:\t2") + return original(path, *args, **kwargs) + + with patch.object(Path, "read_text", changed): + with self.assertRaisesRegex(gate.Refusal, "drift"): + proc.uid_threads(UID) + (directory / "status").unlink() + with self.assertRaises(FileNotFoundError): + proc.uid_threads(UID) + + + +class BenignCommandTests(unittest.TestCase): + def test_parent_death_payload_sleeps_after_broken_pipe(self): + sleeps = [] + + def broken_write(_fd, _data): + raise BrokenPipeError("supervisor closed stdout") + + fake_os = type("FakeOS", (), {"write": staticmethod(broken_write)}) + fake_time = type("FakeTime", (), { + "sleep": staticmethod(sleeps.append), + }) + script = run.benign_command("parent-death")[-1] + + with patch.dict(sys.modules, {"os": fake_os, "time": fake_time}): + with self.assertRaises(BrokenPipeError): + exec(compile(script, "", "exec"), {}) + + self.assertEqual(sleeps, [30]) + self.assertIn("time.sleep(0.2)", run.benign_command("nominal")[-1]) + + +class BarrierTests(unittest.TestCase): + """Sockets locaux et threads inertes ; aucun fork/exec/systemd.""" + + def setUp(self): + self.temp = tempfile.TemporaryDirectory() + self.addCleanup(self.temp.cleanup) + self.path = Path(self.temp.name) / "gate.sock" + + + def peer(self, message, reply=None): + errors = [] + def client(): + try: + with socket.socket(socket.AF_UNIX, socket.SOCK_SEQPACKET) as channel: + channel.settimeout(1) + channel.connect(str(self.path)) + channel.sendall(message) + if reply is not None: + received = channel.recv(4096) + if received != b"RELEASE " + SPEC["nonce"].encode(): + errors.append("wrong release") + channel.sendall(reply) + except BaseException as error: + errors.append(error) + thread = threading.Thread(target=client) + thread.start() + self.addCleanup(lambda: thread.join(timeout=2)) + return thread, errors + + def barrier(self, seconds=1): + barrier = gate.Barrier(self.path, SPEC["nonce"], UID, PID, + time.monotonic() + seconds) + self.addCleanup(barrier.close) + return barrier + + def test_real_channel_peer_credentials_and_single_release(self): + barrier = self.barrier() + message = json.dumps({"nonce": SPEC["nonce"], "pid": PID}).encode() + thread, errors = self.peer(message, b"EXEC_REQUESTED " + SPEC["nonce"].encode()) + self.assertEqual(barrier.accept(), PID) + barrier.release() + with self.assertRaisesRegex(gate.Refusal, "replay"): + barrier.accept() + with self.assertRaisesRegex(gate.Refusal, "seconde"): + barrier.release() + thread.join(timeout=2) + self.assertEqual(errors, []) + + def test_wrong_nonce_and_pid_refuse(self): + messages = ({"nonce": "bad", "pid": PID}, + {"nonce": SPEC["nonce"], "pid": PID + 1}) + for data in messages: + barrier = self.barrier() + thread, _ = self.peer(json.dumps(data).encode()) + with self.assertRaisesRegex(gate.Refusal, "nonce/PID"): + barrier.accept() + thread.join(timeout=2) + barrier.close() + # Remove registered cleanup because close is intentionally idempotent. + + def test_eof_and_accept_timeout(self): + barrier = self.barrier(.05) + with self.assertRaises(TimeoutError): + barrier.accept() + barrier.close() + barrier = self.barrier() + thread, _ = self.peer(b"") + with self.assertRaises((ValueError, gate.Refusal)): + barrier.accept() + thread.join(timeout=2) + + def test_bootstrap_no_exec_on_eof_wrong_nonce_parentdeath_timeout(self): + for action in ("eof", "wrong", "death", "timeout"): + with self.subTest(action=action): + listener = socket.socket(socket.AF_UNIX, socket.SOCK_SEQPACKET) + listener.bind(str(self.path)) + listener.listen(1) + reader, writer = os.pipe() + class Proc: + def identity(self, pid): + return {"pid": PID, "uid": UID, "startticks": 123} + def pidfd(self, pid): + return reader + def server(): + with listener.accept()[0] as channel: + channel.recv(4096) + if action == "wrong": + channel.sendall(b"RELEASE wrong") + elif action == "death": + os.close(writer) + # The pidfd EOF itself proves parent death to bootstrap. + elif action == "timeout": + time.sleep(.05) + thread = threading.Thread(target=server) + thread.start() + with patch.object(gate, "select", wraps=gate.select) as select_module: + if action == "timeout": + select_module.select.return_value = ([], [], []) + with self.assertRaises((gate.Refusal, OSError)): + gate.bootstrap(self.path, SPEC["nonce"], PID, 123, + ["forbidden"], Proc(), + execute=native_spawn_forbidden) + thread.join(timeout=2) + listener.close() + self.path.unlink() + if action != "death": + os.close(writer) + + def test_bootstrap_exec_boundary_is_clean_and_closes_control_fds(self): + listener = socket.socket(socket.AF_UNIX, socket.SOCK_SEQPACKET) + self.addCleanup(listener.close) + listener.bind(str(self.path)) + listener.listen(1) + reader, writer = os.pipe() + self.addCleanup(lambda: os.close(writer)) + observations = [] + errors = [] + + class Proc: + def identity(self, pid): + return {"pid": PID, "uid": UID, "startticks": 123} + def pidfd(self, pid): + return reader + + def server(): + try: + with listener.accept()[0] as channel: + channel.recv(4096) + channel.sendall(b"RELEASE " + SPEC["nonce"].encode()) + observations.append(channel.recv(4096)) + except BaseException as error: + errors.append(error) + + thread = threading.Thread(target=server) + thread.start() + self.addCleanup(lambda: thread.join(timeout=2)) + executed = [] + with patch.object(gate.os, "closerange") as closed, \ + patch.object(gate.os, "chdir") as chdir: + with self.assertRaisesRegex(gate.Refusal, "exec retourné"): + gate.bootstrap(self.path, SPEC["nonce"], PID, 123, ["fixed", "payload"], + Proc(), execute=lambda *args: executed.append(args)) + thread.join(timeout=2) + self.assertEqual(errors, []) + self.assertEqual(executed, [("fixed", ["fixed", "payload"], gate.CLEAN_ENV)]) + self.assertEqual(observations, [b"EXEC_REQUESTED " + SPEC["nonce"].encode()]) + self.assertEqual(closed.call_args.args[0], 3) + chdir.assert_called_once_with("/tmp") + with self.assertRaises(OSError): + os.fstat(reader) + + + +class ManagerTests(unittest.TestCase): + def test_actual_argv_and_environment_with_injected_popen(self): + seen = [] + manager = gate.SystemManager( + UID, popen=lambda argv, **kwargs: seen.append((argv, kwargs))) + manager.create(SPEC, ["fixed-bootstrap"]) + argv, kwargs = seen[0] + self.assertIn("--scope", argv) + self.assertIn("--property=TasksMax=256", argv) + self.assertIn("--property=RuntimeMaxSec=12s", argv) + self.assertIn("--property=TimeoutStopSec=2s", argv) + self.assertIn("--property=KillMode=control-group", argv) + self.assertEqual(argv[-1], "fixed-bootstrap") + self.assertTrue(kwargs["close_fds"]) + self.assertEqual(kwargs["env"]["XDG_RUNTIME_DIR"], f"/run/user/{UID}") + self.assertNotIn("pass_fds", kwargs) + self.assertNotIn("DBUS_SESSION_BUS_ADDRESS", gate.CLEAN_ENV) + + +class ProductionDrainTests(unittest.TestCase): + """Pipes réels à taille minuscule, sans enfant ni exécutable.""" + + def pipes(self, stdout=b"", stderr=b""): + process = FakeProcess() + for name, data in (("stdout", stdout), ("stderr", stderr)): + reader, writer = os.pipe() + os.write(writer, data) + os.close(writer) + stream = os.fdopen(reader, "rb") + setattr(process, name, stream) + self.addCleanup(stream.close) + return process + + def test_exact_bound_and_both_channels(self): + process = self.pipes(b"x" * 16, b"y" * 16) + channels = gate.drain(process, time.monotonic() + .1, limit=16) + self.assertEqual(channels, [b"x" * 16, b"y" * 16]) + self.assertEqual(process.returncode, 0) + + def test_overflow_preserves_bounded_prefix(self): + process = self.pipes(b"x" * 17) + with self.assertRaises(gate.DrainFailure) as context: + gate.drain(process, time.monotonic() + .1, limit=16) + self.assertIn("overflow", str(context.exception)) + self.assertEqual(context.exception.channels[0], b"x" * 16) + + def test_timeout_and_read_failure_are_bounded(self): + process = self.pipes(b"x") + with self.assertRaises(gate.DrainFailure): + gate.drain(process, time.monotonic() - .1) + with patch.object(gate.os, "read", side_effect=OSError("mock read error")): + with self.assertRaisesRegex(gate.DrainFailure, "mock read"): + gate.drain(process, time.monotonic() + .1) + + def test_manager_call_failure_collects_only_its_child(self): + process = self.pipes(b"x" * 17) + manager = gate.SystemManager(UID, popen=lambda *args, **kwargs: process) + with patch.object(gate, "drain", side_effect=OSError("manager drain crash")): + with self.assertRaisesRegex(OSError, "manager drain"): + manager.call(["injected"], time.monotonic() + .2) + self.assertTrue(process.killed) + self.assertTrue(process.stdout.closed) + self.assertTrue(process.stderr.closed) + + +class ElfPreparationTests(unittest.TestCase): + """Analyse ELF de production avec capture passive simulée.""" + + HEADER = ("Class: ELF64\nMachine: Advanced Micro Devices X86-64\n" + "Build ID: abc123\n") + + def test_systemd_direct_runpath_is_recorded_and_not_inherited(self): + import prepare + tool = Path("/usr/bin/systemctl") + library = Path("/usr/lib/systemd/libsystemd-test.so") + libc = Path("/usr/lib/libc.so.6") + data = { + tool: self.HEADER + "(RUNPATH) [/usr/lib/systemd]\n" + "(NEEDED) [libsystemd-test.so]\n", + library: self.HEADER + "(NEEDED) [libc.so.6]\n", + libc: self.HEADER, + } + # A private namesake must not capture a transitive libc edge. + candidates = {library, libc, Path("/usr/lib/systemd/libc.so.6")} + with ( + patch.object(prepare, "capture", side_effect=lambda argv: data[Path(argv[-1])].encode()), + patch.object(Path, "is_file", lambda path: path in candidates), + patch.object(prepare, "trusted_system_path") as trusted, + ): + closure = prepare.elf_closure([tool]) + self.assertEqual(closure[str(tool)]["runpath"], ["/usr/lib/systemd"]) + self.assertEqual(closure[str(library)]["runpath"], []) + self.assertEqual(set(closure), {str(path) for path in data}) + trusted.assert_any_call(Path("/usr/lib/systemd")) + + def test_unapproved_runpaths_and_rpath_are_refused(self): + import prepare + for tag, value in (("RUNPATH", "$ORIGIN"), ("RUNPATH", "relative"), + ("RUNPATH", "/tmp"), + ("RUNPATH", "/usr/lib/systemd:/usr/lib"), + ("RUNPATH", "/usr/lib/systemd:"), + ("RPATH", "/usr/lib/systemd")): + with self.subTest(tag=tag, value=value), patch.object( + prepare, "capture", return_value=( + self.HEADER + f"({tag}) [{value}]\n").encode()): + with self.assertRaises(ValueError): + prepare.elf_closure([Path("/usr/bin/systemctl")]) + with patch.object(prepare, "capture", return_value=( + self.HEADER + "(RUNPATH) [/usr/lib/systemd]\n").encode()): + with self.assertRaises(ValueError): + prepare.elf_closure([Path("/usr/bin/other")]) + + def test_writable_system_component_is_refused(self): + import prepare + with tempfile.TemporaryDirectory() as temporary: + with self.assertRaisesRegex(ValueError, "modifiable"): + prepare.trusted_system_path(Path(temporary)) + + +class PreparationTests(unittest.TestCase): + def test_benign_preparation_uses_no_compiler_and_seals_actual_scripts(self): + import prepare + with tempfile.TemporaryDirectory() as temporary: + lab = Path(temporary) / "fresh" + argv = ["prepare.py", "--lab", str(lab), "--benign"] + with ( + patch.object(sys, "argv", argv), + patch.object(prepare, "capture", side_effect=native_spawn_forbidden), + patch.object(prepare, "elf_closure", return_value={}) as closure, + contextlib.redirect_stdout(io.StringIO()), + ): + prepare.main() + manifest = json.loads((lab / "manifest.json").read_text()) + closure.assert_called_once() + self.assertEqual(manifest["kind"], "benign") + self.assertEqual(manifest["argv"], run.benign_command("nominal")) + self.assertEqual(manifest["scope"], gate.policy(manifest["scope"]["nonce"])) + self.assertIn(str(lab / "fixture/cgroup_gate.py"), manifest["files"]) + self.assertFalse((lab / "ATTEMPT").exists()) + self.assertFalse((lab / "bin/probe").exists()) + + + def test_exact_isolated_passive_cli_creates_no_cache_or_process_payload(self): + import prepare + with tempfile.TemporaryDirectory() as temporary: + lab = Path(temporary) / "fresh" + preparation_argv = ["prepare.py", "--lab", str(lab), "--benign"] + with ( + patch.object(sys, "argv", preparation_argv), + patch.object(prepare, "capture", side_effect=native_spawn_forbidden), + patch.object(prepare, "elf_closure", return_value={}), + contextlib.redirect_stdout(io.StringIO()), + ): + prepare.main() + manifest_hash = run.digest(lab / "manifest.json") + argv = ["/usr/bin/python3", "-I", str(lab / "fixture/run.py"), + "--lab", str(lab), "--manifest-sha256", manifest_hash] + before = sorted(str(path.relative_to(lab)) for path in lab.rglob("*")) + + def inert_only(command, *args, **kwargs): + if command != argv: + return native_spawn_forbidden(command, *args, **kwargs) + return INERT_POPEN(command, *args, **kwargs) + + # This exact source-only Python command is permitted; every other + # executable/argument boundary remains forbidden by the sentinel. + with patch.object(subprocess, "Popen", side_effect=inert_only): + process = subprocess.run(argv, env=run.HOST_ENV, capture_output=True, + timeout=3, check=False) + self.assertEqual(process.returncode, 0, process.stderr.decode()) + self.assertEqual(json.loads(process.stdout)["runtime"], "NOT_RUN") + self.assertEqual(before, sorted(str(path.relative_to(lab)) + for path in lab.rglob("*"))) + self.assertFalse((lab / "ATTEMPT").exists()) + + + +if __name__ == "__main__": + unittest.main() diff --git a/tooling/probes/step20c-egl-fx/test_probe.py b/tooling/probes/step20c-egl-fx/test_probe.py index 29e8cc0..2e224e0 100644 --- a/tooling/probes/step20c-egl-fx/test_probe.py +++ b/tooling/probes/step20c-egl-fx/test_probe.py @@ -13,6 +13,54 @@ from unittest.mock import patch import run +class SystemctlDurationTests(unittest.TestCase): + """Passer la sortie CLI réelle au parseur puis à l'admission de production.""" + + def admission(self, runtime="12s", stop="2s"): + import test_cgroup_admission as fixtures + gate = run.gate.module + fixture = fixtures.AdmissionTests() + fixture.setUp() + self.addCleanup(fixture.doCleanups) + evidence = fixture.preflight() + fixture.cgroups.present = True + properties = { + "Id": fixtures.SPEC["unit"], "Description": fixtures.SPEC["description"], + "InvocationID": "c" * 32, "ControlGroup": fixtures.GROUP, + "Delegate": "yes", "TasksMax": "256", "KillMode": "control-group", + "RuntimeMaxUSec": runtime, "TimeoutStopUSec": stop, + } + # Literal CLI text, independently of POLICY. No systemctl is spawned. + text = "\n".join(f"{key}={value}" for key, value in properties.items() + if value is not None) + manager = gate.SystemManager(fixtures.UID, popen=fixtures.native_spawn_forbidden) + + def captured(argv, deadline): + if "--user" in argv: + return text + return "\n".join(f"{key}={value}" for key, value in fixture.manager.owner.items()) + + with patch.object(manager, "call", side_effect=captured) as capture: + receipt = gate.admit(fixtures.SPEC, evidence, 43, fixture.proc, + fixture.cgroups, manager, 1) + self.addCleanup(os.close, receipt["pidfd"]) + self.assertIn("show", capture.call_args_list[0].args[0]) + return receipt + + def test_rendered_durations_pass_actual_show_parser_and_admission(self): + receipt = self.admission() + self.assertEqual(receipt["properties"]["RuntimeMaxUSec"], "12s") + self.assertEqual(receipt["properties"]["TimeoutStopUSec"], "2s") + + def test_infinite_absent_and_different_durations_are_refused(self): + for runtime, stop in (("infinity", "2s"), (None, "2s"), ("10s", "2s"), + ("12s", "infinity"), ("12s", None), ("12s", "3s"), + ("12000000", "2000000")): + with self.subTest(runtime=runtime, stop=stop): + with self.assertRaisesRegex(run.gate.module.Refusal, "propriétés/budget"): + self.admission(runtime, stop) + + class ProtocolTests(unittest.TestCase): def setUp(self): self.temporary = tempfile.TemporaryDirectory() @@ -23,15 +71,22 @@ class ProtocolTests(unittest.TestCase): (self.lab / "fixture").mkdir() (self.lab / "bin/probe").write_bytes(b"INERT-NOT-ELF") (self.lab / "fixture/run.py").write_text("# inert\n") + (self.lab / "fixture/cgroup_gate.py").write_text("# inert gate\n") self.files = [self.lab / "bin/probe", self.lab / "fixture/run.py", + self.lab / "fixture/cgroup_gate.py", Path("/usr/bin/python3"), + Path("/usr/bin/systemd-run"), Path("/usr/bin/systemctl"), + Path("/usr/bin/env"), Path("/usr/bin/readelf"), Path("/usr/bin/bwrap"), Path("/usr/bin/prlimit")] self.manifest = { - "schema": 1, "lab": str(self.lab), "argv": run.command(self.lab), + "schema": 2, "kind": "egl", "qualification_mode": "nominal", + "scope": run.gate.policy("a" * 32), "lab": str(self.lab), "argv": run.command(self.lab), "environment": run.ENV, "supervisor_environment": run.HOST_ENV, "files": {str(path): {"sha256": run.digest(path), "resolved": str(path.resolve())} for path in self.files}, } + self.manifest["scope_argv"] = run.gate.sealed_scope_argv(self.lab, self.manifest) + def seal(self): (self.lab / "manifest.json").write_text(json.dumps(self.manifest)) return run.digest(self.lab / "manifest.json") @@ -44,7 +99,7 @@ class ProtocolTests(unittest.TestCase): return run.verify(self.lab, seal) def test_valid_identity(self): - self.assertEqual(self.verify()["schema"], 1) + self.assertEqual(self.verify()["schema"], 2) def test_manifest_drift(self): seal = self.seal() @@ -64,6 +119,22 @@ class ProtocolTests(unittest.TestCase): with self.assertRaises(run.Refusal): self.verify(seal) + def test_scope_argv_and_policy_drift(self): + self.manifest["scope_argv"].append("--property=TasksMax=infinity") + with self.assertRaises(run.Refusal): + self.verify() + self.manifest["scope_argv"] = run.gate.sealed_scope_argv(self.lab, self.manifest) + self.manifest["scope"]["properties"] = {"TasksMax": "infinity"} + with self.assertRaises(run.Refusal): + self.verify() + + def test_helper_drift_refuses_before_import(self): + seal = self.seal() + (self.lab / "fixture/cgroup_gate.py").write_text("raise AssertionError('import')") + with patch.object(run._SealedHelper, "__getattr__", side_effect=AssertionError("import")): + with self.assertRaisesRegex(run.Refusal, "helper"): + self.verify(seal) + def test_forbidden_mount_and_devices(self): for device in ("/dev/dri", "/dev/input", "/dev/udmabuf", "/etc", "/sys", "/run/user"): with self.subTest(device=device): @@ -113,7 +184,7 @@ class ProtocolTests(unittest.TestCase): def test_invalid_identity_stops_before_native_boundary(self): with patch.object(sys, "argv", ["run.py", "--lab", str(self.lab), "--manifest-sha256", "bad", "--approve-native-probe"]), \ - patch.object(run, "supervise", side_effect=AssertionError("native boundary")), \ + patch.object(run.gate, "run_scope", side_effect=AssertionError("native boundary")), \ contextlib.redirect_stderr(io.StringIO()): self.assertEqual(run.main(), 20) self.assertFalse((self.lab / "ATTEMPT").exists()) @@ -127,9 +198,12 @@ class ProtocolTests(unittest.TestCase): def test_spawn_failure_consumes_budget(self): arguments = ["run.py", "--lab", str(self.lab), "--manifest-sha256", "test", "--approve-native-probe"] + def spawn_failure(lab, manifest, reserve, journal): + reserve({"mock": True}) + return {"cause": "spawn_failure", "error": "mock spawn failure", "collected": True}, [b"", b""] with patch.object(sys, "argv", arguments), \ patch.object(run, "verify", return_value=self.manifest), \ - patch.object(run, "supervise", side_effect=OSError("mock spawn failure")) as native, \ + patch.object(run.gate, "run_scope", side_effect=spawn_failure) as native, \ contextlib.redirect_stdout(io.StringIO()), contextlib.redirect_stderr(io.StringIO()): self.assertEqual(run.main(), 20) self.assertEqual(run.main(), 20)