feat(lab): add isolated iPhone capture evidence
This commit is contained in:
DCCONSTRUCTIONS
parent
63dd8c8790
commit
19ab973110
|
|
@ -0,0 +1,76 @@
|
|||
# LixelGO / iPhone observation runbook
|
||||
|
||||
This runbook prepares the Mac without full Xcode and defines the first bounded,
|
||||
read-only capture. The accepted decision is ADR 0005. Do not start the iPhone
|
||||
experiment until the exact-profile physical regression in ADR 0004 passes.
|
||||
|
||||
## Current Mac baseline
|
||||
|
||||
- macOS 26.5.1, Apple Silicon;
|
||||
- Xcode Command Line Tools 16.4 are present;
|
||||
- full Xcode and `rvictl` are absent;
|
||||
- built-in `tcpdump` and `uv` are present;
|
||||
- Wireshark/tshark are not required for capture;
|
||||
- the isolated lab uses Python 3.12 and `pymobiledevice3==9.36.0`.
|
||||
|
||||
## Preparation
|
||||
|
||||
```bash
|
||||
uv sync --project plugins/xgrids-k1/lab/iphone-capture --frozen
|
||||
uv run --project plugins/xgrids-k1/lab/iphone-capture --frozen \
|
||||
pymobiledevice3 version
|
||||
```
|
||||
|
||||
Expected version: `9.36.0`.
|
||||
|
||||
Connect one unlocked iPhone using a data cable, approve the Mac accessory prompt,
|
||||
and tap **Trust This Computer**. Do not paste a UDID into Terminal or notes.
|
||||
|
||||
```bash
|
||||
uv run --project plugins/xgrids-k1/lab/iphone-capture --frozen \
|
||||
python plugins/xgrids-k1/lab/iphone-capture/preflight.py
|
||||
```
|
||||
|
||||
## First smoke without K1
|
||||
|
||||
After the iPhone preflight succeeds, run a 10-second idle capture only after the
|
||||
current source milestone is committed or with an explicitly recorded dirty-tree
|
||||
fingerprint:
|
||||
|
||||
```bash
|
||||
uv run --project plugins/xgrids-k1/lab/iphone-capture --frozen \
|
||||
python plugins/xgrids-k1/lab/iphone-capture/capture.py \
|
||||
--duration 10
|
||||
```
|
||||
|
||||
The tool retains the original metadata-bearing `PCAPNG` and creates a compatible
|
||||
classic `PCAP`. Validate the latter with built-in `tcpdump`:
|
||||
|
||||
```bash
|
||||
tcpdump -nn -r sessions/iphone-k1-observation/<session>/captures/iphone-network.pcap -c 1
|
||||
```
|
||||
|
||||
After the capture and operator timeline are both closed, seal the complete
|
||||
session without rewriting the capture manifest:
|
||||
|
||||
```bash
|
||||
uv run python plugins/xgrids-k1/lab/iphone-capture/session_integrity.py \
|
||||
sessions/iphone-k1-observation/<session>
|
||||
```
|
||||
|
||||
The derived private inventory verifies all manifest hashes and also hashes the
|
||||
late operator-event artifact when it exists.
|
||||
|
||||
## Full baseline sequence
|
||||
|
||||
1. Complete the ADR 0004 physical K1 regression without LixelGO.
|
||||
2. Close unrelated iPhone apps and pause unrelated network activity.
|
||||
3. Keep the iPhone on the K1/shared Wi-Fi while USB remains connected to Mac.
|
||||
4. Start a bounded 60–180 second network capture.
|
||||
5. Perform one predefined LixelGO action at a recorded monotonic offset.
|
||||
6. Stop and retain the raw capture; analyze only offline.
|
||||
7. Record endpoints, ports, sizes, timing and correlation before interpreting a
|
||||
payload as a command, image, or model frame.
|
||||
|
||||
The first baseline does not install the Bluetooth logging profile. BLE/HCI is a
|
||||
separate operator-approved experiment.
|
||||
|
|
@ -0,0 +1,76 @@
|
|||
# ADR 0005: Owner-controlled LixelGO/iPhone observation
|
||||
|
||||
- Status: accepted
|
||||
- Date: 2026-07-16
|
||||
- Supersedes: only the unavailable-phone assumption in ADR 0001
|
||||
|
||||
## Context
|
||||
|
||||
ADR 0001 correctly described the original stand: no phone and no LixelGO were
|
||||
available, so they could not be dependencies or test oracles. An
|
||||
owner-controlled iPhone with LixelGO is now available for a new, evidence-only
|
||||
experiment. The previous no-phone results remain valid historical evidence.
|
||||
|
||||
An ordinary third client on the mesh Wi-Fi cannot normally observe unicast
|
||||
traffic between the iPhone and K1. Apple's native Remote Virtual Interface path
|
||||
requires `rvictl`, which is installed by full Xcode and is absent from the
|
||||
current Mac. Installing full Xcode solely for the first capture is unnecessary.
|
||||
|
||||
## Decision
|
||||
|
||||
The XGRIDS plugin may use a pinned, isolated `pymobiledevice3` lab environment
|
||||
to read the iPhone `pcapd` service over a trusted USB connection. This mirrors
|
||||
the iPhone IP stack while the phone remains connected to K1 over Wi-Fi. It is a
|
||||
lab subprocess, not a root Mission Core dependency.
|
||||
|
||||
The experiment is evidence-only:
|
||||
|
||||
- no TLS interception or MITM certificate;
|
||||
- no app automation;
|
||||
- no new BLE or MQTT write;
|
||||
- no router, firmware, device-file, or application mutation;
|
||||
- no camera or command semantic claim until bounded decode and a controlled
|
||||
physical correlation are both present.
|
||||
|
||||
Network capture does not include BLE HCI. Bluetooth capture is a separate gate
|
||||
that requires Apple's diagnostic logging profile, explicit operator approval,
|
||||
and removal of that profile after the experiment.
|
||||
|
||||
All captures are sensitive, mode `0600`, bounded in duration, stored under
|
||||
ignored `sessions/`, hashed, and represented in Git/Ops only by redacted facts.
|
||||
No device ID, credential, SSID, router metadata, or raw packet payload enters
|
||||
normal Git history.
|
||||
|
||||
## Entry gate
|
||||
|
||||
LixelGO/iPhone observation starts only after ADR 0004's post-refactor physical
|
||||
regression succeeds on exact firmware `3.0.2` and direct-LAN:
|
||||
|
||||
1. operator attestation;
|
||||
2. prepare the receiver;
|
||||
3. physical double-click start;
|
||||
4. first real point frame and pose through the new normalizer/viewer path;
|
||||
5. raw evidence finalization;
|
||||
6. operator-confirmed physical stop;
|
||||
7. verification that Mission Core emitted no K1 application command.
|
||||
|
||||
## Consequences
|
||||
|
||||
Full Xcode, `rvictl`, Wireshark, and system/Homebrew changes are not required for
|
||||
the first IP capture. TLS or proprietary encryption may remain opaque; endpoint,
|
||||
port, size, timing, interface, process, and plaintext local protocol evidence are
|
||||
still useful. Any discovered state-changing contract requires a new profile
|
||||
revision and its own acceptance gate before implementation.
|
||||
|
||||
## Outcome recorded on 2026-07-16
|
||||
|
||||
Lab 002 completed the bounded observation. It found a plaintext local MQTT data
|
||||
and control plane plus two RTSP/H.264 camera preview endpoints. Owner-operated
|
||||
LixelGO start and stop were correlated with bounded protobuf action values `1`
|
||||
and `2`, device success responses, physical K1 transitions and start/stop of the
|
||||
high-rate streams. No LixelGO WAN flow was observed in either principal window.
|
||||
|
||||
This outcome updates descriptive evidence only. Mission Core still emits no
|
||||
modeling request: private header/settings construction, save-complete semantics,
|
||||
timeouts and rollback remain unresolved. See
|
||||
`docs/lab/002_LIXELGO_IPHONE_LOCAL_PROTOCOL_20260716.redacted.md`.
|
||||
|
|
@ -0,0 +1,180 @@
|
|||
# Lab 002 — LixelGO/iPhone local K1 protocol observation
|
||||
|
||||
- Date: 2026-07-16
|
||||
- Device scope: one owner-controlled LixelKity K1, exact firmware `3.0.2`
|
||||
- Application scope: owner-operated LixelGO on a trusted iPhone
|
||||
- Capture path: iPhone `pcapd` over USB to the Mac
|
||||
- Result: **GO for protocol-backed camera discovery and start/stop mapping;
|
||||
PAUSE for Mission Core command emission and save-state semantics**
|
||||
|
||||
This report contains no device identifier, address, SSID, credential, project
|
||||
name, room image, or raw packet payload. The sensitive source captures remain
|
||||
under ignored `sessions/` with mode `0600`.
|
||||
|
||||
## Experiment boundary
|
||||
|
||||
The operator performed one clean LixelGO lifecycle and one control/stop window:
|
||||
|
||||
1. K1 was powered to steady-green standby without a physical scan gesture.
|
||||
2. LixelGO established Bluetooth control and requested the K1 Wi-Fi network.
|
||||
3. The operator accepted the iOS Wi-Fi join request.
|
||||
4. LixelGO entered its scan workspace; this navigation alone caused no K1 state
|
||||
transition.
|
||||
5. The operator supplied a private project name and confirmed start.
|
||||
6. K1 performed static initialization and exposed point cloud, pose and camera
|
||||
preview.
|
||||
7. The operator selected left and right camera views.
|
||||
8. The operator opened the stop confirmation and then confirmed it.
|
||||
9. K1 stopped the LiDAR, displayed its saving state and returned to steady-green
|
||||
standby without any physical K1 button press.
|
||||
|
||||
Bluetooth HCI was not captured. The BLE stage is supported by the operator's
|
||||
screen/physical observation, while all protocol findings below start at the
|
||||
iPhone IP stack.
|
||||
|
||||
## Evidence integrity
|
||||
|
||||
The two principal network sessions contain `108,188` and `207,123` packets.
|
||||
Their raw PCAPNG and compatible classic PCAP hashes match the capture manifests,
|
||||
and packet timestamps/order are equivalent across the two representations.
|
||||
|
||||
The original manifests and captures were left byte-for-byte unchanged. A
|
||||
separate private integrity inventory re-hashes the manifests, packet files and
|
||||
late operator-event JSONL. Operator events record sequence, not precise action
|
||||
latency: several were narrated after the corresponding wire event.
|
||||
|
||||
## Network and privacy result
|
||||
|
||||
LixelGO accounted for more than `99.5%` of bytes in each principal capture and
|
||||
communicated with exactly two private local endpoints. No LixelGO flow to a
|
||||
public/WAN address was observed in either window. The first window contained a
|
||||
very small amount of unrelated public traffic from other iPhone processes; the
|
||||
second contained no public traffic at all.
|
||||
|
||||
This is bounded negative evidence for these sessions, not a permanent claim
|
||||
about every application mode or future LixelGO release.
|
||||
|
||||
Observed K1 transports:
|
||||
|
||||
| Transport | Purpose | Result |
|
||||
| --- | --- | --- |
|
||||
| MQTT 3.1.1 over plaintext TCP `1883` | control, status, point cloud, pose | observed |
|
||||
| RTSP 1.0 over TCP `8554` | left/right camera session control | observed |
|
||||
| interleaved RTP/TCP, H.264 PT96, 90 kHz | camera preview media | observed |
|
||||
| HTTP over TCP `80` | three empty project-related setup responses | ancillary only |
|
||||
|
||||
No LixelGO UDP flow was observed.
|
||||
|
||||
## Start and static initialization
|
||||
|
||||
The application start transaction is now physical wire evidence rather than
|
||||
static-analysis evidence:
|
||||
|
||||
- QoS 2 PUBLISH topic: `lixel/application/request/modeling`;
|
||||
- bounded protobuf top-level action field: `1`;
|
||||
- correlated device response: about `1.5 ms` after the request, with the same
|
||||
action and a success result;
|
||||
- first pose: about `23.1 s` after the request;
|
||||
- first modeling/point-cloud reports: about `24.1 s` after the request;
|
||||
- first camera RTSP PLAY: about `25.1 s` after the request.
|
||||
|
||||
The observed 144-byte START protobuf has this bounded top-level wire shape
|
||||
(private strings remain redacted):
|
||||
|
||||
```text
|
||||
f1/wire2 = 130-byte private header/session string
|
||||
f2/wire0 = 1
|
||||
f3/wire2 = 3-byte private string
|
||||
f4/wire0 = 2
|
||||
f5/wire0 = 1
|
||||
```
|
||||
|
||||
`f4` is the decoded value `2`; the occasionally noted `1` describes only the
|
||||
one-byte varint encoding length and is not the field value. The successful
|
||||
response repeats the private 130-byte field, reports `f2=1`, and includes an
|
||||
observed success block in `f15/wire2` (15 bytes). These observations describe
|
||||
the capture but do not establish safe construction rules for a new request.
|
||||
|
||||
The 23–25 second interval matches the operator-observed static initialization.
|
||||
The captured request contains additional private header/settings context. Its
|
||||
complete construction and validation rules remain unresolved and are not
|
||||
committed as a reusable command fixture.
|
||||
|
||||
## Live data plane
|
||||
|
||||
The existing repository decoder successfully decoded distributed samples of
|
||||
both live spatial channels:
|
||||
|
||||
- point cloud: `lixel/application/report/lio_pcl`, about `9.4 Hz`, roughly
|
||||
`2,500–2,650` points per sampled frame;
|
||||
- pose: `lixel/application/report/lio_pose`, about `10 Hz`, finite decoded
|
||||
position/orientation fields.
|
||||
|
||||
During active scanning the aggregate K1-to-phone data plane averaged roughly
|
||||
`1.11 MB/s`, split approximately into `0.25 MB/s` MQTT and `0.86 MB/s` camera
|
||||
RTSP/RTP. The observed peak was about `1.89 MB/s`.
|
||||
|
||||
## Camera preview
|
||||
|
||||
The K1 exposes two independent path-labelled RTSP preview endpoints:
|
||||
|
||||
```text
|
||||
rtsp://{confirmed-device-private-ipv4}:8554/live/chn_left_main
|
||||
rtsp://{confirmed-device-private-ipv4}:8554/live/chn_right_main
|
||||
```
|
||||
|
||||
Each session follows `OPTIONS`, `DESCRIBE`, `SETUP`, `PLAY`; media is H.264 in
|
||||
interleaved RTP over TCP. The observed preview rate was approximately `10 fps`
|
||||
and `6.45–6.59 Mbit/s`.
|
||||
|
||||
Camera selection is not an MQTT command. LixelGO tears down the current RTSP
|
||||
session and opens the other endpoint. In the controlled left-to-right switch,
|
||||
the new RTSP connection began about `0.8 s` after the preceding stream's last
|
||||
packet. RTP arrived immediately after PLAY, but the first new SPS/PPS/IDR group
|
||||
arrived about `0.79 s` later. This explains most of the operator-observed visual
|
||||
delay; exact tap-to-display latency was not measurable from retrospective event
|
||||
markers.
|
||||
|
||||
The capture proves a compressed preview path, not full-resolution raw camera
|
||||
frames, camera calibration, stitched panorama geometry, or optical left/right
|
||||
identity beyond the endpoint labels and operator-selected views.
|
||||
|
||||
## Stop and post-stop boundary
|
||||
|
||||
The remote stop transaction is also verified inside the packet capture:
|
||||
|
||||
- QoS 2 PUBLISH on the same modeling-request topic;
|
||||
- bounded protobuf top-level action field: `2`;
|
||||
- correlated success response after about `33 ms`;
|
||||
- final point-cloud PUBLISH after about `36.4 ms`;
|
||||
- final pose PUBLISH after about `40.8 ms`;
|
||||
- final interleaved video packet after about `54.1 ms`;
|
||||
- point-cloud UNSUBSCRIBE and RTSP TEARDOWN at about `0.95 s`;
|
||||
- status/modeling reports for about another `9 s`;
|
||||
- control MQTT and heartbeat remained reachable for about `31 s` without a
|
||||
recurrence of high-rate point, pose, or video traffic.
|
||||
|
||||
The raw modeling-state value changed immediately after stop. Its enum name is
|
||||
not decoded. The capture does not contain an independently verified
|
||||
`save-complete`/`project-closed` response or the previously observed standby
|
||||
state code. Completed saving and final steady-green standby are therefore
|
||||
operator-corroborated, not packet-semantically proven.
|
||||
|
||||
## Decision
|
||||
|
||||
- **GO:** retain MQTT point cloud and pose as the current live spatial plane.
|
||||
- **GO:** implement a read-only left/right RTSP/H.264 preview receiver behind
|
||||
the XGRIDS plugin boundary.
|
||||
- **GO:** promote start=`1` and stop=`2` from static hints to observed,
|
||||
physical-correlated vendor mappings.
|
||||
- **PAUSE:** keep all Mission Core vendor writes disabled until request headers,
|
||||
settings, acknowledgements, timeout, save completion and rollback are modeled
|
||||
and regression-tested.
|
||||
- **PAUSE:** do not name the post-stop enum or claim durable project persistence
|
||||
from this capture.
|
||||
- **PAUSE:** BLE automation remains governed by the separately reviewed Wi-Fi
|
||||
provisioning profile; no BLE/HCI inference is added here.
|
||||
|
||||
The next smallest implementation step is an offline/then-live read-only camera
|
||||
adapter with bounded buffering, H.264 metadata, explicit left/right selection
|
||||
and no acquisition command publisher.
|
||||
|
|
@ -0,0 +1 @@
|
|||
3.12
|
||||
|
|
@ -0,0 +1,69 @@
|
|||
# iPhone / LixelGO capture lab
|
||||
|
||||
This is an isolated evidence tool for the XGRIDS K1 plugin. It is not a Mission
|
||||
Core runtime dependency and must not be bundled without a separate GPL license
|
||||
review.
|
||||
|
||||
## Install without Xcode
|
||||
|
||||
From the repository root:
|
||||
|
||||
```bash
|
||||
uv sync --project plugins/xgrids-k1/lab/iphone-capture --frozen
|
||||
```
|
||||
|
||||
The environment is created under the lab directory. It pins
|
||||
`pymobiledevice3==9.36.0`; it does not modify the system Python, Homebrew, the
|
||||
Mission Core `.venv`, or the root lockfile.
|
||||
|
||||
## USB preflight
|
||||
|
||||
Connect exactly one iPhone with a data-capable USB cable, unlock it, confirm the
|
||||
Mac accessory prompt, and tap **Trust This Computer** on the iPhone. Then run:
|
||||
|
||||
```bash
|
||||
uv run --project plugins/xgrids-k1/lab/iphone-capture --frozen \
|
||||
python plugins/xgrids-k1/lab/iphone-capture/preflight.py
|
||||
```
|
||||
|
||||
The preflight prints only a device count. It never prints or stores the iPhone
|
||||
UDID or serial number.
|
||||
|
||||
## Bounded network capture
|
||||
|
||||
Run this only after the post-refactor exact-profile K1 regression in ADR 0004:
|
||||
|
||||
```bash
|
||||
uv run --project plugins/xgrids-k1/lab/iphone-capture --frozen \
|
||||
python plugins/xgrids-k1/lab/iphone-capture/capture.py \
|
||||
--duration 120
|
||||
```
|
||||
|
||||
The capture uses the iPhone `pcapd` service over USB, suppresses packet hex on
|
||||
stdout, and writes private artifacts under ignored
|
||||
`sessions/iphone-k1-observation/`. It is intentionally bounded to 5–600 seconds.
|
||||
Each successful run retains the original metadata-bearing `PCAPNG` and creates a
|
||||
classic Ethernet `PCAP` that the macOS bundled `tcpdump` can read. The redacted
|
||||
manifest records the validated packet count and hashes both artifacts.
|
||||
The default refuses a dirty repository; `--allow-dirty` records a source-tree
|
||||
fingerprint when an explicitly approved lab run cannot wait for a commit.
|
||||
|
||||
Operator events may be recorded while the capture is active and can finish
|
||||
after the bounded network process exits. Preserve the original capture manifest
|
||||
and create a separate integrity inventory after the operator timeline is closed:
|
||||
|
||||
```bash
|
||||
uv run python plugins/xgrids-k1/lab/iphone-capture/session_integrity.py \
|
||||
sessions/iphone-k1-observation/<session>
|
||||
```
|
||||
|
||||
The sealer re-hashes every manifest artifact, adds the manifest and optional
|
||||
`operator-events.private.jsonl`, and writes an ignored mode-`0600`
|
||||
`integrity.private.json`. It never edits the original manifest or raw capture.
|
||||
|
||||
Do not add `--bluetooth` unless Apple's Bluetooth diagnostic logging profile is
|
||||
already installed on the iPhone and the operator explicitly approved that
|
||||
configuration change. The script never installs the profile.
|
||||
|
||||
Wireshark is optional and is not required to record PCAP/PCAPNG. Full Xcode is
|
||||
also not required for this path.
|
||||
|
|
@ -0,0 +1,313 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import hashlib
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import secrets
|
||||
import struct
|
||||
import subprocess
|
||||
import time
|
||||
from datetime import UTC, datetime
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
from pcapng import FileScanner
|
||||
from pcapng.blocks import EnhancedPacket
|
||||
from preflight import (
|
||||
pymobiledevice3_binary,
|
||||
pymobiledevice3_version,
|
||||
usb_device_count,
|
||||
)
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parents[4]
|
||||
SESSION_ROOT = REPO_ROOT / "sessions" / "iphone-k1-observation"
|
||||
SAFE_LABEL = re.compile(r"^[a-z0-9][a-z0-9-]{0,31}$")
|
||||
|
||||
|
||||
def utc_now() -> str:
|
||||
return datetime.now(UTC).isoformat()
|
||||
|
||||
|
||||
def sha256_file(path: Path) -> str:
|
||||
digest = hashlib.sha256()
|
||||
with path.open("rb") as stream:
|
||||
for chunk in iter(lambda: stream.read(1024 * 1024), b""):
|
||||
digest.update(chunk)
|
||||
return digest.hexdigest()
|
||||
|
||||
|
||||
def git_output(*args: str) -> bytes:
|
||||
completed = subprocess.run(
|
||||
["git", *args],
|
||||
cwd=REPO_ROOT,
|
||||
check=True,
|
||||
capture_output=True,
|
||||
)
|
||||
return completed.stdout
|
||||
|
||||
|
||||
def source_snapshot() -> dict[str, Any]:
|
||||
revision = git_output("rev-parse", "HEAD").decode().strip()
|
||||
status = git_output("status", "--porcelain=v1", "--untracked-files=all")
|
||||
dirty = bool(status.strip())
|
||||
digest = hashlib.sha256()
|
||||
digest.update(revision.encode())
|
||||
digest.update(b"\0")
|
||||
digest.update(status)
|
||||
if dirty:
|
||||
digest.update(git_output("diff", "--binary", "HEAD"))
|
||||
untracked = git_output("ls-files", "--others", "--exclude-standard", "-z")
|
||||
for raw_name in filter(None, untracked.split(b"\0")):
|
||||
relative = raw_name.decode("utf-8", errors="surrogateescape")
|
||||
path = REPO_ROOT / relative
|
||||
digest.update(raw_name)
|
||||
digest.update(b"\0")
|
||||
if path.is_file():
|
||||
digest.update(sha256_file(path).encode())
|
||||
return {
|
||||
"git_revision": revision,
|
||||
"source_tree_dirty": dirty,
|
||||
"source_tree_fingerprint_sha256": digest.hexdigest(),
|
||||
}
|
||||
|
||||
|
||||
def terminate(process: subprocess.Popen[bytes]) -> int:
|
||||
if process.poll() is None:
|
||||
process.terminate()
|
||||
try:
|
||||
process.wait(timeout=10)
|
||||
except subprocess.TimeoutExpired:
|
||||
process.kill()
|
||||
process.wait(timeout=5)
|
||||
return int(process.returncode or 0)
|
||||
|
||||
|
||||
def artifact_record(path: Path, kind: str, session_dir: Path) -> dict[str, Any] | None:
|
||||
if not path.is_file() or path.stat().st_size == 0:
|
||||
return None
|
||||
path.chmod(0o600)
|
||||
return {
|
||||
"kind": kind,
|
||||
"relative_path": str(path.relative_to(session_dir)),
|
||||
"size_bytes": path.stat().st_size,
|
||||
"sha256": sha256_file(path),
|
||||
}
|
||||
|
||||
|
||||
def convert_pcapng_to_pcap(source: Path, destination: Path) -> int:
|
||||
"""Create a classic Ethernet PCAP that macOS's bundled tcpdump can read."""
|
||||
packet_count = 0
|
||||
with source.open("rb") as source_stream, destination.open("xb") as destination_stream:
|
||||
destination_stream.write(
|
||||
struct.pack(
|
||||
"<IHHIIII",
|
||||
0xA1B2C3D4,
|
||||
2,
|
||||
4,
|
||||
0,
|
||||
0,
|
||||
262_144,
|
||||
1,
|
||||
)
|
||||
)
|
||||
for block in FileScanner(source_stream):
|
||||
if not isinstance(block, EnhancedPacket):
|
||||
continue
|
||||
packet_data = bytes(block.packet_data)
|
||||
captured_length = int(block.captured_len)
|
||||
packet_length = int(block.packet_len)
|
||||
if captured_length != len(packet_data) or packet_length < captured_length:
|
||||
raise ValueError("invalid enhanced-packet lengths")
|
||||
|
||||
timestamp = float(block.timestamp)
|
||||
seconds = int(timestamp)
|
||||
microseconds = int(round((timestamp - seconds) * 1_000_000))
|
||||
if microseconds >= 1_000_000:
|
||||
seconds += 1
|
||||
microseconds -= 1_000_000
|
||||
destination_stream.write(
|
||||
struct.pack(
|
||||
"<IIII",
|
||||
seconds,
|
||||
microseconds,
|
||||
captured_length,
|
||||
packet_length,
|
||||
)
|
||||
)
|
||||
destination_stream.write(packet_data)
|
||||
packet_count += 1
|
||||
destination.chmod(0o600)
|
||||
return packet_count
|
||||
|
||||
|
||||
def parse_args() -> argparse.Namespace:
|
||||
parser = argparse.ArgumentParser(
|
||||
description="Bounded, redacted USB iPhone capture for the Mission Core K1 lab"
|
||||
)
|
||||
parser.add_argument("--duration", type=int, default=120, help="capture seconds (5..600)")
|
||||
parser.add_argument("--label", default="lixelgo-observe", help="safe session label")
|
||||
parser.add_argument("--process", help="optional iOS process filter for network capture")
|
||||
parser.add_argument("--interface", help="optional iOS interface filter for network capture")
|
||||
parser.add_argument(
|
||||
"--bluetooth",
|
||||
action="store_true",
|
||||
help="also capture iPhone Bluetooth HCI; requires Apple's logging profile",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--acknowledge-ios-bluetooth-profile",
|
||||
action="store_true",
|
||||
help="confirm the operator already installed Apple's Bluetooth logging profile",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--allow-dirty",
|
||||
action="store_true",
|
||||
help="record an explicit dirty-tree fingerprint instead of refusing the capture",
|
||||
)
|
||||
return parser.parse_args()
|
||||
|
||||
|
||||
def main() -> int:
|
||||
args = parse_args()
|
||||
if not 5 <= args.duration <= 600:
|
||||
raise SystemExit("--duration must be between 5 and 600 seconds")
|
||||
if not SAFE_LABEL.fullmatch(args.label):
|
||||
raise SystemExit("--label must use lowercase ASCII letters, digits, and hyphens")
|
||||
if args.bluetooth and not args.acknowledge_ios_bluetooth_profile:
|
||||
raise SystemExit(
|
||||
"Bluetooth capture requires --acknowledge-ios-bluetooth-profile; "
|
||||
"the script never installs the profile"
|
||||
)
|
||||
|
||||
os.umask(0o077)
|
||||
binary = pymobiledevice3_binary()
|
||||
if usb_device_count(binary) != 1:
|
||||
raise SystemExit("exactly one trusted USB iPhone must be connected")
|
||||
snapshot = source_snapshot()
|
||||
if snapshot["source_tree_dirty"] and not args.allow_dirty:
|
||||
raise SystemExit(
|
||||
"repository is dirty; commit the milestone or pass --allow-dirty "
|
||||
"to record its fingerprint"
|
||||
)
|
||||
|
||||
session_id = (
|
||||
datetime.now(UTC).strftime("%Y%m%dT%H%M%SZ") + f"_{args.label}_{secrets.token_hex(2)}"
|
||||
)
|
||||
session_dir = SESSION_ROOT / session_id
|
||||
capture_dir = session_dir / "captures"
|
||||
capture_dir.mkdir(parents=True, mode=0o700)
|
||||
network_pcapng_path = capture_dir / "iphone-network.pcapng"
|
||||
network_pcap_path = capture_dir / "iphone-network.pcap"
|
||||
bluetooth_path = capture_dir / "iphone-bluetooth.pcapng"
|
||||
stderr_path = session_dir / "capture.stderr.log"
|
||||
manifest_path = session_dir / "manifest.redacted.json"
|
||||
|
||||
manifest: dict[str, Any] = {
|
||||
"schema": "missioncore.xgrids-k1/iphone-observation-manifest/v1alpha1",
|
||||
"session_id": session_id,
|
||||
"status": "starting",
|
||||
"started_at_utc": utc_now(),
|
||||
"started_monotonic_ns": time.monotonic_ns(),
|
||||
"requested_duration_seconds": args.duration,
|
||||
"capture_backend": "pymobiledevice3-9.36.0/com.apple.pcapd",
|
||||
"pymobiledevice3_version": pymobiledevice3_version(binary),
|
||||
"network_capture": True,
|
||||
"bluetooth_capture": args.bluetooth,
|
||||
"network_process_filter": args.process,
|
||||
"network_interface_filter": args.interface,
|
||||
"device_identity_recorded": False,
|
||||
"credentials_recorded_in_manifest": False,
|
||||
"source": snapshot,
|
||||
"artifacts": [],
|
||||
}
|
||||
manifest_path.write_text(json.dumps(manifest, indent=2) + "\n", encoding="utf-8")
|
||||
manifest_path.chmod(0o600)
|
||||
|
||||
network_command = [binary, "--no-color", "pcap", "--out", str(network_pcapng_path)]
|
||||
if args.process:
|
||||
network_command.extend(["--process", args.process])
|
||||
if args.interface:
|
||||
network_command.extend(["--interface", args.interface])
|
||||
commands = [("network", network_command)]
|
||||
if args.bluetooth:
|
||||
commands.append(
|
||||
(
|
||||
"bluetooth",
|
||||
[binary, "--no-color", "btlogger", str(bluetooth_path), "--format", "pcapng"],
|
||||
)
|
||||
)
|
||||
|
||||
processes: list[tuple[str, subprocess.Popen[bytes]]] = []
|
||||
exit_codes: dict[str, int] = {}
|
||||
interrupted = False
|
||||
failure: str | None = None
|
||||
with stderr_path.open("wb") as error_log:
|
||||
stderr_path.chmod(0o600)
|
||||
try:
|
||||
for name, command in commands:
|
||||
process = subprocess.Popen(
|
||||
command,
|
||||
cwd=REPO_ROOT,
|
||||
stdout=subprocess.DEVNULL,
|
||||
stderr=error_log,
|
||||
)
|
||||
processes.append((name, process))
|
||||
time.sleep(1)
|
||||
early = [name for name, process in processes if process.poll() is not None]
|
||||
if early:
|
||||
raise RuntimeError(f"capture service exited early: {', '.join(early)}")
|
||||
|
||||
deadline = time.monotonic() + args.duration
|
||||
while time.monotonic() < deadline:
|
||||
ended = [name for name, process in processes if process.poll() is not None]
|
||||
if ended:
|
||||
raise RuntimeError(f"capture service stopped unexpectedly: {', '.join(ended)}")
|
||||
time.sleep(min(0.25, deadline - time.monotonic()))
|
||||
except KeyboardInterrupt:
|
||||
interrupted = True
|
||||
except RuntimeError as exc:
|
||||
failure = str(exc)
|
||||
finally:
|
||||
for name, process in processes:
|
||||
exit_codes[name] = terminate(process)
|
||||
|
||||
network_packet_count: int | None = None
|
||||
if failure is None:
|
||||
try:
|
||||
network_packet_count = convert_pcapng_to_pcap(
|
||||
network_pcapng_path,
|
||||
network_pcap_path,
|
||||
)
|
||||
if network_packet_count == 0:
|
||||
failure = "network capture validation found no packets"
|
||||
except Exception as exc: # noqa: BLE001 - third-party parser errors vary by version
|
||||
failure = f"network capture validation failed: {type(exc).__name__}"
|
||||
|
||||
artifacts = [
|
||||
artifact_record(network_pcapng_path, "iphone-network-pcapng-raw", session_dir),
|
||||
artifact_record(network_pcap_path, "iphone-network-pcap-compatible", session_dir),
|
||||
artifact_record(bluetooth_path, "iphone-bluetooth-hci-pcapng", session_dir),
|
||||
artifact_record(stderr_path, "capture-stderr", session_dir),
|
||||
]
|
||||
manifest.update(
|
||||
{
|
||||
"status": "failed" if failure else "completed",
|
||||
"interrupted_by_operator": interrupted,
|
||||
"failure": failure,
|
||||
"ended_at_utc": utc_now(),
|
||||
"ended_monotonic_ns": time.monotonic_ns(),
|
||||
"process_exit_codes": exit_codes,
|
||||
"network_packet_count": network_packet_count,
|
||||
"artifacts": [item for item in artifacts if item is not None],
|
||||
}
|
||||
)
|
||||
manifest_path.write_text(json.dumps(manifest, indent=2) + "\n", encoding="utf-8")
|
||||
manifest_path.chmod(0o600)
|
||||
print(f"session: {session_dir}")
|
||||
print(f"status: {manifest['status']}")
|
||||
return 1 if failure else 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
|
|
@ -0,0 +1,76 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
def pymobiledevice3_binary() -> str:
|
||||
sibling = Path(sys.executable).with_name("pymobiledevice3")
|
||||
if sibling.is_file():
|
||||
return str(sibling)
|
||||
discovered = shutil.which("pymobiledevice3")
|
||||
if discovered is None:
|
||||
raise RuntimeError("pymobiledevice3 is not installed in this environment")
|
||||
return discovered
|
||||
|
||||
|
||||
def pymobiledevice3_version(binary: str) -> str:
|
||||
completed = subprocess.run(
|
||||
[binary, "--no-color", "version"],
|
||||
check=False,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=20,
|
||||
)
|
||||
if completed.returncode != 0:
|
||||
raise RuntimeError("pymobiledevice3 version check failed")
|
||||
return completed.stdout.strip()
|
||||
|
||||
|
||||
def usb_device_count(binary: str) -> int:
|
||||
completed = subprocess.run(
|
||||
[binary, "--no-color", "usbmux", "list", "--usb"],
|
||||
check=False,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=20,
|
||||
)
|
||||
if completed.returncode != 0:
|
||||
raise RuntimeError(
|
||||
"USB discovery failed; unlock the iPhone and confirm Trust This Computer"
|
||||
)
|
||||
try:
|
||||
payload = json.loads(completed.stdout)
|
||||
except json.JSONDecodeError as exc:
|
||||
raise RuntimeError("USB discovery returned an unexpected response") from exc
|
||||
if not isinstance(payload, list):
|
||||
raise RuntimeError("USB discovery returned an unexpected response")
|
||||
return len(payload)
|
||||
|
||||
|
||||
def main() -> int:
|
||||
try:
|
||||
binary = pymobiledevice3_binary()
|
||||
version = pymobiledevice3_version(binary)
|
||||
count = usb_device_count(binary)
|
||||
except (OSError, RuntimeError, subprocess.SubprocessError) as exc:
|
||||
print(f"BLOCKED: {exc}")
|
||||
return 2
|
||||
|
||||
print(f"pymobiledevice3: {version}")
|
||||
print(f"USB iPhone devices: {count}")
|
||||
if count == 0:
|
||||
print("BLOCKED: connect one unlocked iPhone with a data cable and confirm Trust")
|
||||
return 2
|
||||
if count > 1:
|
||||
print("BLOCKED: leave exactly one trusted iPhone connected for a redacted capture")
|
||||
return 2
|
||||
print("READY: one trusted USB iPhone is available; no device identifier was printed")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
|
|
@ -0,0 +1,12 @@
|
|||
[project]
|
||||
name = "mission-core-ios-capture-lab"
|
||||
version = "0.1.0"
|
||||
description = "Isolated read-only iPhone packet-capture tooling for Mission Core lab work"
|
||||
requires-python = ">=3.12,<3.13"
|
||||
dependencies = [
|
||||
"pymobiledevice3==9.36.0",
|
||||
"python-pcapng==2.1.1",
|
||||
]
|
||||
|
||||
[tool.uv]
|
||||
package = false
|
||||
|
|
@ -0,0 +1,169 @@
|
|||
"""Seal one completed iPhone capture without rewriting its original manifest.
|
||||
|
||||
The capture manifest is written when the bounded packet capture ends. Operator
|
||||
events may be appended after that point, so they belong in a separate derived
|
||||
integrity inventory rather than in a retroactively edited capture manifest.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import hashlib
|
||||
import json
|
||||
import os
|
||||
from datetime import UTC, datetime
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
INTEGRITY_SCHEMA = "missioncore.xgrids-k1/iphone-session-integrity/v1alpha1"
|
||||
CAPTURE_MANIFEST = "manifest.redacted.json"
|
||||
INTEGRITY_INVENTORY = "integrity.private.json"
|
||||
OPTIONAL_SESSION_ARTIFACTS = {
|
||||
"capture.stderr.log": "capture-stderr",
|
||||
"operator-events.private.jsonl": "operator-events",
|
||||
}
|
||||
|
||||
|
||||
class SessionIntegrityError(ValueError):
|
||||
"""The session cannot be safely verified or sealed."""
|
||||
|
||||
|
||||
def sha256_file(path: Path) -> str:
|
||||
digest = hashlib.sha256()
|
||||
with path.open("rb") as stream:
|
||||
for chunk in iter(lambda: stream.read(1024 * 1024), b""):
|
||||
digest.update(chunk)
|
||||
return digest.hexdigest()
|
||||
|
||||
|
||||
def _load_object(path: Path) -> dict[str, Any]:
|
||||
try:
|
||||
value = json.loads(path.read_text(encoding="utf-8"))
|
||||
except (OSError, json.JSONDecodeError) as exc:
|
||||
raise SessionIntegrityError(f"cannot read JSON object: {path.name}") from exc
|
||||
if not isinstance(value, dict):
|
||||
raise SessionIntegrityError(f"JSON root must be an object: {path.name}")
|
||||
return value
|
||||
|
||||
|
||||
def _safe_artifact_path(session_dir: Path, relative_path: object) -> Path:
|
||||
if not isinstance(relative_path, str) or not relative_path:
|
||||
raise SessionIntegrityError("artifact relative_path must be a non-empty string")
|
||||
relative = Path(relative_path)
|
||||
if relative.is_absolute() or ".." in relative.parts:
|
||||
raise SessionIntegrityError("artifact path must stay inside the session")
|
||||
path = session_dir / relative
|
||||
if path.is_symlink() or not path.is_file():
|
||||
raise SessionIntegrityError(f"artifact is missing or is a symlink: {relative_path}")
|
||||
return path
|
||||
|
||||
|
||||
def _record(path: Path, *, kind: str, session_dir: Path) -> dict[str, Any]:
|
||||
before = path.stat()
|
||||
digest = sha256_file(path)
|
||||
after = path.stat()
|
||||
if (before.st_size, before.st_mtime_ns) != (after.st_size, after.st_mtime_ns):
|
||||
raise SessionIntegrityError(f"artifact changed while hashing: {path.name}")
|
||||
return {
|
||||
"kind": kind,
|
||||
"relative_path": str(path.relative_to(session_dir)),
|
||||
"size_bytes": after.st_size,
|
||||
"sha256": digest,
|
||||
"mode_octal": f"{after.st_mode & 0o777:04o}",
|
||||
}
|
||||
|
||||
|
||||
def build_inventory(session_dir: Path) -> dict[str, Any]:
|
||||
"""Verify a completed capture and return a derived integrity inventory."""
|
||||
session_dir = session_dir.expanduser().resolve()
|
||||
manifest_path = session_dir / CAPTURE_MANIFEST
|
||||
if manifest_path.is_symlink() or not manifest_path.is_file():
|
||||
raise SessionIntegrityError(f"missing {CAPTURE_MANIFEST}")
|
||||
|
||||
manifest = _load_object(manifest_path)
|
||||
if manifest.get("status") != "completed":
|
||||
raise SessionIntegrityError("only a completed capture can be sealed")
|
||||
session_id = manifest.get("session_id")
|
||||
if not isinstance(session_id, str) or session_id != session_dir.name:
|
||||
raise SessionIntegrityError("manifest session_id does not match the directory")
|
||||
artifacts = manifest.get("artifacts")
|
||||
if not isinstance(artifacts, list) or not artifacts:
|
||||
raise SessionIntegrityError("capture manifest has no artifacts")
|
||||
|
||||
verified: list[dict[str, Any]] = []
|
||||
for item in artifacts:
|
||||
if not isinstance(item, dict):
|
||||
raise SessionIntegrityError("manifest artifact must be an object")
|
||||
path = _safe_artifact_path(session_dir, item.get("relative_path"))
|
||||
actual = _record(
|
||||
path,
|
||||
kind=str(item.get("kind") or "capture-manifest-artifact"),
|
||||
session_dir=session_dir,
|
||||
)
|
||||
if item.get("size_bytes") != actual["size_bytes"]:
|
||||
raise SessionIntegrityError(f"artifact size mismatch: {actual['relative_path']}")
|
||||
if item.get("sha256") != actual["sha256"]:
|
||||
raise SessionIntegrityError(f"artifact hash mismatch: {actual['relative_path']}")
|
||||
verified.append(actual)
|
||||
|
||||
derived = [_record(manifest_path, kind="capture-manifest", session_dir=session_dir)]
|
||||
for relative_name, kind in OPTIONAL_SESSION_ARTIFACTS.items():
|
||||
path = session_dir / relative_name
|
||||
if path.exists():
|
||||
if path.is_symlink() or not path.is_file():
|
||||
raise SessionIntegrityError(
|
||||
f"optional artifact is not a regular file: {relative_name}"
|
||||
)
|
||||
derived.append(_record(path, kind=kind, session_dir=session_dir))
|
||||
|
||||
return {
|
||||
"schema": INTEGRITY_SCHEMA,
|
||||
"session_id": session_id,
|
||||
"sealed_at_utc": datetime.now(UTC).isoformat(),
|
||||
"capture_manifest_preserved": True,
|
||||
"verified_manifest_artifacts": verified,
|
||||
"derived_session_artifacts": derived,
|
||||
}
|
||||
|
||||
|
||||
def seal_session(session_dir: Path, *, overwrite: bool = False) -> Path:
|
||||
"""Write an ignored mode-0600 inventory next to the immutable capture."""
|
||||
os.umask(0o077)
|
||||
resolved = session_dir.expanduser().resolve()
|
||||
destination = resolved / INTEGRITY_INVENTORY
|
||||
if destination.exists() and not overwrite:
|
||||
raise SessionIntegrityError(f"{INTEGRITY_INVENTORY} already exists")
|
||||
inventory = build_inventory(resolved)
|
||||
temporary = resolved / f".{INTEGRITY_INVENTORY}.tmp"
|
||||
temporary.write_text(json.dumps(inventory, indent=2) + "\n", encoding="utf-8")
|
||||
temporary.chmod(0o600)
|
||||
temporary.replace(destination)
|
||||
destination.chmod(0o600)
|
||||
return destination
|
||||
|
||||
|
||||
def parse_args() -> argparse.Namespace:
|
||||
parser = argparse.ArgumentParser(
|
||||
description="Verify and seal one completed Mission Core iPhone capture session"
|
||||
)
|
||||
parser.add_argument("session", type=Path, help="completed ignored session directory")
|
||||
parser.add_argument(
|
||||
"--overwrite",
|
||||
action="store_true",
|
||||
help="replace only the derived inventory; never changes the capture manifest",
|
||||
)
|
||||
return parser.parse_args()
|
||||
|
||||
|
||||
def main() -> int:
|
||||
args = parse_args()
|
||||
try:
|
||||
destination = seal_session(args.session, overwrite=args.overwrite)
|
||||
except SessionIntegrityError as exc:
|
||||
raise SystemExit(str(exc)) from exc
|
||||
print(f"sealed: {destination}")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
File diff suppressed because it is too large
Load Diff
|
|
@ -0,0 +1,107 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import importlib.util
|
||||
import json
|
||||
from pathlib import Path
|
||||
from types import ModuleType
|
||||
|
||||
import pytest
|
||||
|
||||
REPOSITORY_ROOT = Path(__file__).parents[1]
|
||||
INTEGRITY_PATH = (
|
||||
REPOSITORY_ROOT / "plugins" / "xgrids-k1" / "lab" / "iphone-capture" / "session_integrity.py"
|
||||
)
|
||||
|
||||
|
||||
def _load_module() -> ModuleType:
|
||||
spec = importlib.util.spec_from_file_location(
|
||||
"iphone_capture_session_integrity", INTEGRITY_PATH
|
||||
)
|
||||
assert spec is not None
|
||||
assert spec.loader is not None
|
||||
module = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(module)
|
||||
return module
|
||||
|
||||
|
||||
INTEGRITY = _load_module()
|
||||
|
||||
|
||||
def _sha256(value: bytes) -> str:
|
||||
return hashlib.sha256(value).hexdigest()
|
||||
|
||||
|
||||
def _session(tmp_path: Path) -> tuple[Path, Path]:
|
||||
session = tmp_path / "20260716T000000Z_test_abcd"
|
||||
capture_dir = session / "captures"
|
||||
capture_dir.mkdir(parents=True)
|
||||
capture = capture_dir / "iphone-network.pcap"
|
||||
capture_bytes = b"synthetic-private-capture"
|
||||
capture.write_bytes(capture_bytes)
|
||||
manifest = {
|
||||
"session_id": session.name,
|
||||
"status": "completed",
|
||||
"artifacts": [
|
||||
{
|
||||
"kind": "iphone-network-pcap-compatible",
|
||||
"relative_path": "captures/iphone-network.pcap",
|
||||
"size_bytes": len(capture_bytes),
|
||||
"sha256": _sha256(capture_bytes),
|
||||
}
|
||||
],
|
||||
}
|
||||
(session / "manifest.redacted.json").write_text(json.dumps(manifest), encoding="utf-8")
|
||||
(session / "capture.stderr.log").write_bytes(b"")
|
||||
events = session / "operator-events.private.jsonl"
|
||||
events.write_text('{"sequence":1}\n', encoding="utf-8")
|
||||
return session, capture
|
||||
|
||||
|
||||
def test_seal_session_preserves_manifest_and_hashes_late_operator_events(tmp_path: Path) -> None:
|
||||
session, _ = _session(tmp_path)
|
||||
manifest = session / "manifest.redacted.json"
|
||||
before = manifest.read_bytes()
|
||||
|
||||
destination = INTEGRITY.seal_session(session)
|
||||
inventory = json.loads(destination.read_text(encoding="utf-8"))
|
||||
|
||||
assert manifest.read_bytes() == before
|
||||
assert inventory["capture_manifest_preserved"] is True
|
||||
assert inventory["verified_manifest_artifacts"][0]["relative_path"] == (
|
||||
"captures/iphone-network.pcap"
|
||||
)
|
||||
derived = {item["kind"]: item for item in inventory["derived_session_artifacts"]}
|
||||
assert derived["capture-manifest"]["sha256"] == _sha256(before)
|
||||
assert derived["capture-stderr"]["size_bytes"] == 0
|
||||
assert derived["operator-events"]["relative_path"] == "operator-events.private.jsonl"
|
||||
assert destination.stat().st_mode & 0o777 == 0o600
|
||||
|
||||
|
||||
def test_seal_session_rejects_capture_tampering(tmp_path: Path) -> None:
|
||||
session, capture = _session(tmp_path)
|
||||
capture.write_bytes(b"tampered")
|
||||
|
||||
with pytest.raises(INTEGRITY.SessionIntegrityError, match="size mismatch"):
|
||||
INTEGRITY.seal_session(session)
|
||||
|
||||
|
||||
def test_seal_session_refuses_to_replace_inventory_without_explicit_flag(
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
session, _ = _session(tmp_path)
|
||||
INTEGRITY.seal_session(session)
|
||||
|
||||
with pytest.raises(INTEGRITY.SessionIntegrityError, match="already exists"):
|
||||
INTEGRITY.seal_session(session)
|
||||
|
||||
|
||||
def test_seal_session_rejects_manifest_path_escape(tmp_path: Path) -> None:
|
||||
session, _ = _session(tmp_path)
|
||||
manifest_path = session / "manifest.redacted.json"
|
||||
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
|
||||
manifest["artifacts"][0]["relative_path"] = "../outside.pcap"
|
||||
manifest_path.write_text(json.dumps(manifest), encoding="utf-8")
|
||||
|
||||
with pytest.raises(INTEGRITY.SessionIntegrityError, match="stay inside"):
|
||||
INTEGRITY.seal_session(session)
|
||||
Loading…
Reference in New Issue