Compare commits
1 Commits
us/US07-03
...
us/US06-06
| Author | SHA1 | Date | |
|---|---|---|---|
| db4c362b91 |
@@ -80,8 +80,7 @@ are the authoritative backlog.
|
||||
uncertain. Work only on the claimed story and its generated feature branch.
|
||||
4. Read the entire issue, linked specification, dependencies, and acceptance criteria.
|
||||
Reconcile them with the concept before designing or changing code.
|
||||
5. Inspect the legacy CLI donors — frozen in `legacy_cli_archive/` since US07-01,
|
||||
with their ledger — before replacing applicable behavior. Update the donor
|
||||
5. Inspect the legacy CLI donors before replacing applicable behavior. Update the donor
|
||||
ledger and characterization tests required by the story.
|
||||
6. Implement every acceptance criterion and its automated tests.
|
||||
7. Run story-specific tests and the accumulated regression suite required by the epic.
|
||||
|
||||
76
README.md
76
README.md
@@ -17,25 +17,6 @@ python -m photo_pipeline serve # start the API + static review UI (127.0.0.1
|
||||
Configuration comes from `PHOTO_PIPELINE_*` environment variables (see
|
||||
`photo_pipeline/config.py`); secrets are referenced, never logged.
|
||||
|
||||
### API access (US07-02)
|
||||
|
||||
The app listens on loopback, so its attacker is another page in the same browser.
|
||||
Every `/api/v1` route except `health/live`, `health/ready`, and `session` requires
|
||||
the application session, and every mutation requires its CSRF token as well:
|
||||
|
||||
```bash
|
||||
BASE=http://127.0.0.1:8000
|
||||
TOKEN=$(curl -sc /tmp/pp.jar $BASE/api/v1/session | python -c 'import json,sys; print(json.load(sys.stdin)["csrf_token"])')
|
||||
curl -sb /tmp/pp.jar -H "X-CSRF-Token: $TOKEN" -X POST $BASE/api/v1/albums/proposals -d '{}' -H 'Content-Type: application/json'
|
||||
```
|
||||
|
||||
The session is per server process — restarting `serve` invalidates it, and the
|
||||
browser client re-bootstraps by itself. Requests are also refused when the `Host` is
|
||||
not a loopback name (DNS rebinding), when `Origin` is any other origin, when
|
||||
`Sec-Fetch-Site` says the request came from another site (an `<img>` pointed at a
|
||||
thumbnail), or when the body exceeds `PHOTO_PIPELINE_MAX_REQUEST_BYTES`. There is no
|
||||
CORS middleware at all, so no other origin can read a response.
|
||||
|
||||
## Testing
|
||||
|
||||
One offline command runs the whole suite (unit, integration, and browser
|
||||
@@ -227,60 +208,3 @@ work_item/scripts/python -m pytest -m phase_f -q
|
||||
offline browsing, restore, collision, keyboard confirmation, and reload).
|
||||
|
||||
Phases A–E remain green in the full run above.
|
||||
|
||||
## Media and metadata hardening (US07-03)
|
||||
|
||||
Every pixel the application reads goes through `photo_pipeline/imaging.py`: the
|
||||
declared dimensions are checked before anything is decoded, Pillow's
|
||||
decompression-bomb warning is treated as a refusal, JPEG decodes near the requested
|
||||
size, and each decoder failure becomes one of two typed errors. A damaged file is a
|
||||
per-item error with a persisted code, never a failed scan or a dead worker.
|
||||
|
||||
Every metadata stage ends with an EXIF checkpoint (`services/exif_checkpoint.py`):
|
||||
snapshot, write the owned keywords, read back, prove the owned fields landed and that
|
||||
nothing else moved, refresh the file hash. A field the stage does not own that
|
||||
changed anyway makes the checkpoint `divergent` — recorded in `exif_projections`,
|
||||
shown in the review queue, never repaired behind the user's back, and not counted as
|
||||
verified, so upload stays blocked.
|
||||
|
||||
The golden corpus that proves all of it is generated, not committed:
|
||||
`tests/fixtures/media_corpus.py` declares every format, orientation, profile,
|
||||
damage, and metadata case with its expected outcome, and the suite regenerates it
|
||||
twice to prove it does not drift.
|
||||
|
||||
```bash
|
||||
work_item/scripts/python -m pytest tests/integration/test_media_hardening.py tests/integration/test_exif_checkpoints.py -q
|
||||
```
|
||||
|
||||
## Legacy CLI archive
|
||||
|
||||
The command-line tools this application was extracted from are frozen in
|
||||
`legacy_cli_archive/` (US07-01): the original sources, their docs, the dependency
|
||||
lock they were last verified against, schema notes, a redacted sample
|
||||
configuration, the donor ledger, and a checksum for every file.
|
||||
|
||||
```bash
|
||||
cd legacy_cli_archive && shasum -a 256 -c CHECKSUMS.sha256 # verify the archive
|
||||
work_item/scripts/python -m pytest tests/unit/test_legacy_archive.py -q # lint it
|
||||
```
|
||||
|
||||
They are reference material and rollback evidence only. No module under
|
||||
`photo_pipeline/` imports or executes them, the archive is not on the application's
|
||||
import path, and `tests/unit/test_legacy_archive.py` enforces that along with the
|
||||
checksums and the redaction. Only the two suites that compare *against* the donors —
|
||||
`tests/characterization/` and `tests/integration/test_safety_parity.py` — put the
|
||||
archived sources on `sys.path`.
|
||||
|
||||
The last path-keyed state they owned, `nsfw_scores.csv`, is imported once and then
|
||||
left alone:
|
||||
|
||||
```bash
|
||||
work_item/scripts/python -m photo_pipeline import-legacy-scores /path/to/nsfw_scores.csv --dry-run
|
||||
```
|
||||
|
||||
The import writes scored-but-unreviewed `safety_reviews` rows onto stable asset ids,
|
||||
never invents an asset for an unknown path, never overwrites a human decision, and
|
||||
writes a reconciliation report to the data directory saying exactly what it did.
|
||||
`legacy_cli_archive/donor_ledger.yaml` records every migrated behavior with its
|
||||
target, the tests that pin the donor, the tests that prove the replacement, and each
|
||||
intentional delta; rows still marked `pending` name the story that will resolve them.
|
||||
|
||||
@@ -12,15 +12,7 @@
|
||||
# (if anything) carries over
|
||||
# Every row needs either `tests` (existing test IDs, module::function) or
|
||||
# `pending_story` (the backlog story that will characterize/deliver it).
|
||||
# status: characterized — donor behavior pinned by characterization tests
|
||||
# resolved — replacement shipped; `parity` names the tests that prove
|
||||
# it, and `delta` states every intentional difference
|
||||
# pending — not migrated yet; `pending_story` says which story will
|
||||
# parity: test ids (path::function) in any suite, proving the replacement
|
||||
# delta: what the replacement deliberately does differently, or not at all
|
||||
#
|
||||
# Archived by US07-01: the sources referenced below now live beside this file in
|
||||
# src/ and are frozen (see README.md). Nothing in photo_pipeline imports them.
|
||||
# status: characterized | pending
|
||||
|
||||
rows:
|
||||
# ── photo_analyzer.py ──────────────────────────────────────────────────────
|
||||
@@ -88,12 +80,8 @@ rows:
|
||||
Copying a primary's analysis into variant rows survives, but keyed by
|
||||
asset_id and recorded as stage state instead of raw row copies.
|
||||
target: photo_pipeline/services/duplicates.py
|
||||
parity:
|
||||
- tests/integration/test_duplicate_engine.py::test_exact_copies_form_auto_decided_cluster
|
||||
- tests/integration/test_duplicate_engine.py::test_perceptual_variant_is_review_only
|
||||
delta: >
|
||||
The donor propagated variant links implicitly while writing rows; the replacement makes the canonical link a reviewable, reversible cluster decision, so a propagated link can always be undone.
|
||||
status: resolved
|
||||
pending_story: US01-04
|
||||
status: pending
|
||||
|
||||
- id: pa-hashing
|
||||
area: hashing
|
||||
@@ -153,12 +141,8 @@ rows:
|
||||
classification: replace
|
||||
rationale: Console report; superseded by the duplicate-review API/UI (US01-06).
|
||||
target: photo_pipeline/api/routes + frontend duplicate review
|
||||
parity:
|
||||
- tests/integration/test_review_api.py::test_clusters_list_and_detail
|
||||
- tests/e2e/test_review_ui.py::test_fuzzy_decision_requires_confirmation
|
||||
delta: >
|
||||
The text listing became the paged cluster API and the comparison UI; no textual report is produced.
|
||||
status: resolved
|
||||
pending_story: US01-06
|
||||
status: pending
|
||||
|
||||
- id: pa-reconcile
|
||||
area: database
|
||||
@@ -218,7 +202,7 @@ rows:
|
||||
RGB-normalize (drops alpha, converts HEIC), LANCZOS resize to 2048px
|
||||
long-edge, JPEG q85 base64 — the provider-input contract. Truncated-image
|
||||
tolerance (ImageFile.LOAD_TRUNCATED_IMAGES) carries with it.
|
||||
target: photo_pipeline/services/analysis.py
|
||||
target: photo_pipeline/integrations/vision.py
|
||||
tests:
|
||||
- test_pa_imaging::test_prepare_image_small_passthrough_jpeg
|
||||
- test_pa_imaging::test_prepare_image_resizes_to_max_long_edge
|
||||
@@ -246,12 +230,9 @@ rows:
|
||||
response validation, 429/503 retry with exponential backoff. Prompt and
|
||||
model/config version must be persisted per analysis_runs. Characterized
|
||||
against a deterministic fake provider when the analysis service is ported.
|
||||
target: photo_pipeline/services/analysis.py
|
||||
parity:
|
||||
- tests/integration/test_safety_analysis.py::test_provider_called_only_for_confirmed_sfw
|
||||
delta: >
|
||||
The prompt and response schema carry over; the provider is an injected adapter so the privacy gate is testable, and results are keyed to asset ids rather than paths.
|
||||
status: resolved
|
||||
target: photo_pipeline/integrations/vision.py
|
||||
pending_story: US02-06
|
||||
status: pending
|
||||
|
||||
- id: pa-throttle
|
||||
area: logging
|
||||
@@ -260,8 +241,8 @@ rows:
|
||||
rationale: >
|
||||
Rolling throttle window + persistent throttle_events.jsonl + RPD day
|
||||
counter become job metrics/events on the durable job model.
|
||||
target: photo_pipeline/services/jobs.py + photo_pipeline/services/analysis.py
|
||||
pending_story: US07-04
|
||||
target: photo_pipeline/jobs/coordinator.py
|
||||
pending_story: US02-02
|
||||
status: pending
|
||||
|
||||
- id: pa-nsfw-filter
|
||||
@@ -331,12 +312,8 @@ rows:
|
||||
checked between items, double-SIGINT force quit — becomes the durable
|
||||
JobRunner worker loop with the same drain-and-resume semantics.
|
||||
target: photo_pipeline/jobs/worker.py
|
||||
parity:
|
||||
- tests/integration/test_worker.py::test_worker_processes_all_items
|
||||
- tests/integration/test_worker.py::test_cooperative_cancellation_leaves_items_resumable
|
||||
delta: >
|
||||
The in-process folder loop with SIGINT handling became durable jobs claimed by a worker: cancellation is a persisted request, not a signal, and an interrupted run resumes from the database instead of restarting.
|
||||
status: resolved
|
||||
pending_story: US02-02
|
||||
status: pending
|
||||
|
||||
- id: pa-ui-terminal
|
||||
area: ui
|
||||
@@ -394,12 +371,8 @@ rows:
|
||||
the JSONL history logger become structured JSON logging with job_id/
|
||||
asset_id and job_events rows; per-photo history maps to job events.
|
||||
target: photo_pipeline structured logging + jobs/job_events
|
||||
parity:
|
||||
- tests/integration/test_jobs.py::test_enqueue_persists_items_and_event
|
||||
- tests/integration/test_jobs_sse.py::test_sse_streams_all_events_then_closes
|
||||
delta: >
|
||||
The JSONL history file and rich console handler are replaced by structured JSON logs plus durable job_events; the browser reads events over SSE rather than tailing a file.
|
||||
status: resolved
|
||||
pending_story: US02-02
|
||||
status: pending
|
||||
|
||||
- id: pa-balance
|
||||
area: vision
|
||||
@@ -409,11 +382,8 @@ rows:
|
||||
Provider balance/quota probes (report 'unsupported' on providers without
|
||||
the endpoint). Network-bound; characterized against the fake provider.
|
||||
target: photo_pipeline/services/analysis.py
|
||||
parity:
|
||||
- tests/integration/test_safety_analysis.py::test_provider_called_only_for_confirmed_sfw
|
||||
delta: >
|
||||
Not carried over: balance/quota polling was provider-specific (Gemini/OpenAI billing endpoints) and key-scoped. Cost reporting, when a story asks for it, comes from the per-run usage recorded with each analysis result rather than from a vendor endpoint.
|
||||
status: resolved
|
||||
pending_story: US02-06
|
||||
status: pending
|
||||
|
||||
- id: pa-cli
|
||||
area: configuration
|
||||
@@ -423,13 +393,9 @@ rows:
|
||||
argparse surface is superseded by the API; flags map to job configs
|
||||
(documented in WEBAPP_CONCEPT.md §8 parity table). Transitional CLI calls
|
||||
the shared services until archival (E07).
|
||||
target: photo_pipeline/__main__.py (serve | worker | migrate | import-legacy-scores) + /api/v1
|
||||
parity:
|
||||
- tests/integration/test_app_lifecycle.py::test_restart_preserves_data_and_reruns_migrations
|
||||
- tests/unit/test_legacy_archive.py::test_production_code_never_imports_an_archived_module
|
||||
delta: >
|
||||
The argparse surface is not reproduced. Every flag that drove work became an API command or a job configuration; the CLI keeps only what an application needs to be operated (serve, worker, migrate) plus the one-off legacy CSV import.
|
||||
status: resolved
|
||||
target: photo_pipeline/api + transitional CLI
|
||||
pending_story: US07-01
|
||||
status: pending
|
||||
|
||||
# ── nsfwtag/ ───────────────────────────────────────────────────────────────
|
||||
- id: nt-discovery
|
||||
@@ -456,17 +422,9 @@ rows:
|
||||
nsfw_scores.csv stops being the source of truth (concept: DB state).
|
||||
Format characterized (4-decimal scores, bad rows dropped) because the
|
||||
existing CSV must migrate into assets.safety_score.
|
||||
target: photo_pipeline/services/legacy_import.py (one-off import into safety_reviews)
|
||||
target: photo_pipeline/repositories (safety), CSV import in US01-02 migration
|
||||
tests: [test_nsfwtag::test_score_cache_roundtrip_and_tolerance]
|
||||
parity:
|
||||
- tests/integration/test_legacy_import.py::test_scores_are_imported_onto_asset_identity
|
||||
- tests/integration/test_legacy_import.py::test_a_reviewed_asset_is_never_overwritten_by_the_csv
|
||||
delta: >
|
||||
The CSV is no longer read at runtime at all: it is imported once into
|
||||
scored-but-unreviewed safety_reviews rows and left on disk untouched. A path
|
||||
that matches nothing is reported, never turned into an asset, and a human
|
||||
decision always outranks an imported score.
|
||||
status: resolved
|
||||
status: characterized
|
||||
|
||||
- id: nt-score-model
|
||||
area: nsfw
|
||||
@@ -480,13 +438,6 @@ rows:
|
||||
inference path needs the local model + deterministic fake.
|
||||
target: photo_pipeline/integrations/nsfw_model.py
|
||||
tests: [test_nsfwtag::test_score_images_cache_hit_skips_model]
|
||||
delta: >
|
||||
The donor set Pillow's process-global ImageFile.LOAD_TRUNCATED_IMAGES so a
|
||||
partially downloaded file still scored. Here the same process also hashes
|
||||
files and renders previews, and those must keep refusing a truncated file
|
||||
rather than silently working on half of one; scoring opens images through
|
||||
the bounded photo_pipeline.imaging door instead and skips the ones it cannot
|
||||
read, leaving them unscored and visibly undecided (US07-03).
|
||||
status: characterized
|
||||
|
||||
- id: nt-exif-keyword
|
||||
@@ -534,11 +485,8 @@ rows:
|
||||
Newline-list bulk tagging (nsfw_confirmed.txt flow) is superseded by DB
|
||||
review decisions; the existing list is a one-time migration input.
|
||||
target: photo_pipeline/services/safety.py (decision import in US01-02)
|
||||
parity:
|
||||
- tests/integration/test_safety_parity.py::test_extracted_marks_and_partition_match_donor
|
||||
delta: >
|
||||
Bulk keyword application from a file list is replaced by decisions against asset ids; the EXIF write itself is the extracted, read-back-verified one.
|
||||
status: resolved
|
||||
pending_story: US01-02
|
||||
status: pending
|
||||
|
||||
- id: nt-ui
|
||||
area: ui
|
||||
@@ -550,11 +498,8 @@ rows:
|
||||
threshold/score review flow, lightbox and keyboard model are the frontend
|
||||
donor for the Safety view (preserved per concept §10; ported in US02-01).
|
||||
target: photo_pipeline/api + frontend Safety view
|
||||
parity:
|
||||
- tests/e2e/test_workflow_views.py::test_safety_review_decide_persists_across_reload
|
||||
delta: >
|
||||
The stdlib review server is replaced by the API plus the Safety view; the donor's layout, thresholds, and keyboard flow carry over, its embedded HTML generation does not.
|
||||
status: resolved
|
||||
pending_story: US02-01
|
||||
status: pending
|
||||
|
||||
- id: nt-bench
|
||||
area: nsfw
|
||||
@@ -564,11 +509,8 @@ rows:
|
||||
Dev-only model benchmark; archived without webapp replacement (recorded
|
||||
basis of the AdamCodd model choice). No production caller.
|
||||
target: none (archive as reference)
|
||||
parity:
|
||||
- tests/unit/test_legacy_archive.py::test_every_archived_source_matches_its_checksum
|
||||
delta: >
|
||||
No replacement: a dev-only benchmark whose result (the AdamCodd model choice) is already recorded. Kept in the archive as the basis of that choice.
|
||||
status: resolved
|
||||
pending_story: US07-01
|
||||
status: pending
|
||||
|
||||
# ── webapp/ ────────────────────────────────────────────────────────────────
|
||||
- id: wa-query-search
|
||||
@@ -637,13 +579,9 @@ rows:
|
||||
Subprocess-driving-the-CLI job control is superseded by durable DB jobs
|
||||
with a worker process. Two ideas carry over: progress derived from DB
|
||||
counts (not job-private state) and single-mutating-job enforcement.
|
||||
target: photo_pipeline/services/jobs.py + photo_pipeline/jobs/worker.py
|
||||
parity:
|
||||
- tests/integration/test_jobs.py::test_idempotency_key_returns_same_job
|
||||
- tests/integration/test_worker.py::test_handler_failure_fails_the_job
|
||||
delta: >
|
||||
Subprocess supervision of a CLI is replaced by durable jobs in the same process family: there is no subprocess to supervise, and progress is persisted rather than scraped from stdout.
|
||||
status: resolved
|
||||
target: photo_pipeline/jobs/coordinator.py
|
||||
pending_story: US02-02
|
||||
status: pending
|
||||
|
||||
- id: wa-server
|
||||
area: ui
|
||||
@@ -655,9 +593,5 @@ rows:
|
||||
browser. analyzer.html + page.py design (dark OLED tokens, Library/
|
||||
Analyze/Stats views) is frontend donor material per concept §10.
|
||||
target: photo_pipeline/api/app.py + frontend
|
||||
parity:
|
||||
- tests/e2e/test_frontend_shell.py::test_shell_loads_assets_without_console_or_network_errors
|
||||
- tests/e2e/test_frontend_shell.py::test_deep_link_and_reload_restore_view_and_filters
|
||||
delta: >
|
||||
The stdlib HTTP routes become the versioned FastAPI API and the static frontend shell; Python no longer interpolates HTML.
|
||||
status: resolved
|
||||
pending_story: US02-05
|
||||
status: pending
|
||||
@@ -2,43 +2,14 @@
|
||||
// cancellation. Every method accepts an optional { signal } from cancellable().
|
||||
export const BASE = "/api/v1";
|
||||
|
||||
// The API refuses every request without the session cookie, and every mutation
|
||||
// without this token echoed back. The token is readable only same-origin, which is
|
||||
// what makes it proof that the caller is this app and not another page.
|
||||
let csrfToken = null;
|
||||
|
||||
async function session() {
|
||||
if (csrfToken === null) {
|
||||
const response = await fetch(BASE + "/session", { credentials: "same-origin" });
|
||||
const body = await response.json().catch(() => null);
|
||||
csrfToken = (body && body.csrf_token) || null;
|
||||
}
|
||||
return csrfToken || "";
|
||||
}
|
||||
|
||||
async function send(path, { signal, ...options }) {
|
||||
return fetch(BASE + path, {
|
||||
credentials: "same-origin",
|
||||
signal,
|
||||
...options,
|
||||
headers: {
|
||||
"Content-Type": "application/json",
|
||||
"X-CSRF-Token": await session(),
|
||||
...(options.headers || {}),
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
async function request(path, { signal, ...options } = {}) {
|
||||
let response;
|
||||
try {
|
||||
response = await send(path, { signal, ...options });
|
||||
// A restarted server issues a new session; re-bootstrap once rather than
|
||||
// stranding an open tab on 401.
|
||||
if (response.status === 401) {
|
||||
csrfToken = null;
|
||||
response = await send(path, { signal, ...options });
|
||||
}
|
||||
response = await fetch(BASE + path, {
|
||||
headers: { "Content-Type": "application/json" },
|
||||
signal,
|
||||
...options,
|
||||
});
|
||||
} catch (error) {
|
||||
// A caller-cancelled fetch is not a failure; tag it so views can ignore it.
|
||||
if (error.name === "AbortError") {
|
||||
|
||||
@@ -29,10 +29,6 @@ function jsonResponse(status, body) {
|
||||
const tick = (ms = 0) => new Promise((r) => setTimeout(r, ms));
|
||||
|
||||
async function run() {
|
||||
// The client fetches its CSRF token once, lazily (US07-02). Do that against the
|
||||
// real server first, so the stubbed fetch below only ever sees the call under test.
|
||||
await api.workflow().catch(() => {});
|
||||
|
||||
// ── store ────────────────────────────────────────────────────────────────
|
||||
{
|
||||
const store = createStore({ n: 0 });
|
||||
|
||||
@@ -1,22 +0,0 @@
|
||||
d098bbde13d2ebc872ca781e244cc1e48b97cc551fedaca3251d6a5278a15234 src/compare_models.py
|
||||
7d0f68cb95fbd6603e4c558620098b46929ebdb6fd91a598bbf84f26eea35e6c src/nsfw_tag.py
|
||||
35329c53570e215cef15f59429c1a5251a0448b90522b0910ab9d68efe1bc307 src/nsfwtag/__init__.py
|
||||
7f52b55e4f3b77eda3657d7cd2272c7cca8422100e241671de495603c1126ee2 src/nsfwtag/__main__.py
|
||||
6f5a97114e0d87d272ce22d065dc31fa0dd72ff8590d11ad14cb1c1f486d339c src/nsfwtag/bench.py
|
||||
29498fac1d73ba2b7420ffe1ffe49a87f684ec299e2c3b9c6e14e63e95643ba4 src/nsfwtag/exif.py
|
||||
934e82c402813ebf503e5a20eb03d5103f84df13117bc4dd1fd95deaa01ab263 src/nsfwtag/README.md
|
||||
68125e6184c7e4d2a5b0153f2155675753ab0b1329d1dbc4933d4769492ec4be src/nsfwtag/review.html
|
||||
a597eab74803dd31452b5d70abf7d6d6320eb17a61596c891b009da15d624e98 src/nsfwtag/scoring.py
|
||||
674969a18e58ee511a2abf574a875db92f2f524769b1b614d0d3662ab879a709 src/nsfwtag/server.py
|
||||
24a7c8d029da6e97d46b110a9fe9dbb3900127f8ae3c142f634fdd91d6c45243 src/nsfwtag/webapp.py
|
||||
2c2ea558f1b9095c1150f078ea29c8e0f180cc045a9d1097b83f61bfb145295b src/photo_analyzer.py
|
||||
5fb7ce2f977a17da1f501c45d7985328318ae14298a3f347eab46a6e3f02aab3 src/test_dedup.py
|
||||
1673dc76cc60aba56a5b065d9bc9f5dda00342cf2a79d000a3f74686fb7cffec src/test_nsfw_skip.py
|
||||
55aa82f348e0e90be0163ba6aa278e5fc8e445996e5335774917e5ef59ad3563 src/webapp/__init__.py
|
||||
ba4342bd0175591a2121f063f14a78a602d5679bcfcf24aa14c8d4b8cd5da1a2 src/webapp/__main__.py
|
||||
a60f24035a989909467778f888d62854c1afc2702f47b8d3cca32bf3cf04d5cf src/webapp/analyzer.html
|
||||
018bcc2f770d716444b456db58d6b4a41800138a98598b3c27ba503ab22d939e src/webapp/page.py
|
||||
5f599b107b2b117ca118b6fbdec5e5786a9ab4eea424aa9af1bfea7bf87cebab src/webapp/query.py
|
||||
875cce697caa02717c85a2707ba935c46dd9ace18428fd4abb13105b7a8e24d9 src/webapp/README.md
|
||||
af7d0d72d245b4bbb1e0e30f9708239697543212d4171defd1bfe26dd24a4f05 src/webapp/runner.py
|
||||
cddf0555b06fe7e4daa14309baaaf91130f8506922cb0ffb6507d76cbe39add4 src/webapp/server.py
|
||||
@@ -1,80 +0,0 @@
|
||||
# Legacy CLI archive (US07-01)
|
||||
|
||||
Frozen, read-only sources of the command-line tools this application was extracted
|
||||
from. They are **reference material and rollback evidence** — provenance for
|
||||
behavior that now lives in `photo_pipeline/`, and the only way to answer "what did
|
||||
the original actually do?" once the replacement has drifted.
|
||||
|
||||
> **Nothing here is production code.** No module under `photo_pipeline/` imports or
|
||||
> executes anything in this directory, and this directory is not on the application's
|
||||
> import path. `tests/unit/test_legacy_archive.py` enforces both, along with the
|
||||
> checksums and the redaction below.
|
||||
|
||||
## What is here
|
||||
|
||||
| Path | Role |
|
||||
|---|---|
|
||||
| `src/photo_analyzer.py` | the analysis CLI: discovery, hashing/dedup, vision analysis, EXIF writing, SQLite schema, album naming |
|
||||
| `src/nsfwtag/` | NSFW scoring, EXIF safety keywords, and the review server (`__init__` 1.1.0) |
|
||||
| `src/webapp/` | the stdlib review web app: FTS search, stats, subprocess runner, HTML shell (`__init__` 0.1.0) |
|
||||
| `src/nsfw_tag.py` | thin backwards-compatible entry point for `nsfwtag` |
|
||||
| `src/compare_models.py` | dev-only model comparison script |
|
||||
| `src/test_dedup.py`, `src/test_nsfw_skip.py` | the CLIs' own standalone self-checks (never pytest suites) |
|
||||
| `donor_ledger.yaml` | the donor ledger: every migrated behavior, its target, its tests, and every intentional delta |
|
||||
| `requirements-lock.txt` | the dependency versions the frozen sources were last verified against |
|
||||
| `photo_analyzer.env.sample` | the CLI's configuration surface, with every value replaced by a placeholder |
|
||||
| `CHECKSUMS.sha256` | SHA-256 of every archived source file |
|
||||
|
||||
`photo_analyzer.py` carries no `__version__`; its identity is its checksum, recorded
|
||||
in `CHECKSUMS.sha256` and taken at commit `9b7ee6b` (the merge of US06-06, the last
|
||||
commit before archival).
|
||||
|
||||
## Verifying the archive
|
||||
|
||||
```bash
|
||||
cd legacy_cli_archive && shasum -a 256 -c CHECKSUMS.sha256
|
||||
```
|
||||
|
||||
Any edit to an archived source must be accompanied by a regenerated checksum file
|
||||
and a note here explaining why a *frozen* archive changed — the normal answer being
|
||||
that it should not.
|
||||
|
||||
## Schema notes
|
||||
|
||||
`photo_analyzer.py` owned a path-keyed SQLite database (`SCHEMA`, near the top of the
|
||||
file):
|
||||
|
||||
- `photos(id, path UNIQUE, status, phash, file_sha1, dup_of, description, tags,
|
||||
people_count, setting, time_of_day, season, mood, location_hint, approx_year,
|
||||
raw_response, error_message, analyzed_at, exif_written_at)`;
|
||||
- `photos_fts` — an FTS5 external-content index over `path, description, tags, mood,
|
||||
location_hint`, kept in sync by insert/update/delete triggers;
|
||||
- late columns (`phash`, `file_sha1`, `dup_of`) were added by an in-code
|
||||
`_migrate_schema()` rather than a migration tool, and their indexes are created
|
||||
only after the `ALTER`.
|
||||
|
||||
The replacement keeps the same analysis fields but re-keys everything to a stable
|
||||
`assets.id` (Alembic migrations `0001`…), because a path is not an identity: the
|
||||
donor's `path UNIQUE` is exactly what broke on every move and rename.
|
||||
|
||||
`nsfwtag` kept its safety scores outside the database in `nsfw_scores.csv`
|
||||
(`path,nsfw_score`, four decimals, unreadable rows dropped). That file is no longer a
|
||||
source of truth; `photo_pipeline/services/legacy_import.py` imports it into
|
||||
`assets.safety_score` and reports exactly what matched, what did not, and why.
|
||||
|
||||
## Redaction
|
||||
|
||||
The archive contains no credentials. `photo_analyzer.env.sample` documents the
|
||||
configuration surface (`LLM_API_KEY`, `LLM_BASE_URL`, `LLM_MODEL`, and the tuning
|
||||
variables) with placeholder values only; the CLI itself never contained a key, it
|
||||
read one from `photo_analyzer.env` or the environment. No `.env`, database, log, CSV,
|
||||
or photo from the author's library is archived.
|
||||
|
||||
## Why these tools were retired
|
||||
|
||||
Each behavior's fate is recorded per row in `donor_ledger.yaml`: `reuse`, `extract`,
|
||||
`refactor`, or `replace`, with the target module, the characterization tests that
|
||||
pinned the donor's behavior, the parity tests the replacement passes, and — where the
|
||||
replacement deliberately does something else or nothing at all — a `delta` saying so.
|
||||
Rows still marked `pending` name the backlog story that will resolve them; they are
|
||||
the honest list of what has *not* been carried over yet.
|
||||
@@ -1,18 +0,0 @@
|
||||
# Sample configuration for the archived photo_analyzer.py CLI (US07-01).
|
||||
#
|
||||
# REDACTED: no value below is real. Each line names a variable the CLI read and
|
||||
# what belongs there; the placeholders are deliberately not key-shaped, so this
|
||||
# file can never be mistaken for — or scanned as — a credential.
|
||||
|
||||
LLM_API_KEY=<paste your provider key here>
|
||||
LLM_BASE_URL=<provider base url, e.g. the OpenAI-compatible Gemini endpoint>
|
||||
LLM_MODEL=<model id, e.g. a Gemini Flash release>
|
||||
|
||||
# Optional tuning the CLI read from the same file:
|
||||
PHASH_THRESHOLD=8
|
||||
MAX_WORKERS=4
|
||||
RETRY_ATTEMPTS=3
|
||||
RPD_LIMIT=0
|
||||
|
||||
# The CLI ignored its own shipped placeholder (a literal "sk-REPLACE..." string)
|
||||
# until it was replaced, and a shell variable always won over this file.
|
||||
@@ -1,20 +0,0 @@
|
||||
# Final dependency lock of the archived CLIs (US07-01).
|
||||
#
|
||||
# These are the versions present in the environment the archive was taken from —
|
||||
# what the frozen sources were last verified against by the characterization
|
||||
# suite. Restoring a donor for forensics means pinning these, not "latest".
|
||||
#
|
||||
# Python 3.14.6
|
||||
|
||||
openai==3.0.0 # photo_analyzer: OpenAI-compatible vision client
|
||||
numpy==2.4.6 # photo_analyzer, nsfwtag: pixel work
|
||||
Pillow==12.3.0 # photo_analyzer, nsfwtag: decode/resize
|
||||
rich==15.0.0 # photo_analyzer: console output
|
||||
scipy==1.18.0 # photo_analyzer: perceptual-hash DCT
|
||||
PyYAML==6.0.3 # tooling that reads the donor ledger
|
||||
|
||||
# NSFW inference (nsfwtag/scoring.py, nsfwtag/bench.py) was never installed in the
|
||||
# archiving environment; the model stack is recorded here from the sources so a
|
||||
# forensic run can reproduce it, not from a resolved lock:
|
||||
# torch, transformers, timm — AdamCodd/vit-base-nsfw-detector (see nsfwtag/README.md)
|
||||
# exiftool is an external binary, not a Python package.
|
||||
@@ -1,42 +0,0 @@
|
||||
"""Durable EXIF projections per asset and stage (US07-03).
|
||||
|
||||
Revision ID: 0015_exif_projections
|
||||
Revises: 0014_restore_plans
|
||||
Create Date: 2026-08-16
|
||||
|
||||
The concept's ``exif_projections`` table, added at the point it earns its keep: a
|
||||
checkpoint that finds a field it does not own changed must be able to say so after
|
||||
a restart. ``state`` is verified | divergent | failed, and only ``verified`` counts
|
||||
as a completed metadata stage.
|
||||
"""
|
||||
|
||||
import sqlalchemy as sa
|
||||
from alembic import op
|
||||
|
||||
revision = "0015_exif_projections"
|
||||
down_revision = "0014_restore_plans"
|
||||
branch_labels = None
|
||||
depends_on = None
|
||||
|
||||
|
||||
def upgrade() -> None:
|
||||
op.create_table(
|
||||
"exif_projections",
|
||||
sa.Column("asset_id", sa.String(), sa.ForeignKey("assets.id"), primary_key=True),
|
||||
sa.Column("stage", sa.String(), primary_key=True), # safety | analysis
|
||||
sa.Column("id", sa.String(), nullable=False),
|
||||
sa.Column("projection_version", sa.Integer(), nullable=False, server_default="0"),
|
||||
# What the stage asked for: {"add": [...], "remove": [...]}.
|
||||
sa.Column("desired_json", sa.String(), nullable=True),
|
||||
# Fields outside the stage's ownership that did not survive the write.
|
||||
sa.Column("divergent_fields", sa.String(), nullable=True),
|
||||
sa.Column("result_file_sha256", sa.String(), nullable=True),
|
||||
sa.Column("state", sa.String(), nullable=False),
|
||||
sa.Column("error_code", sa.String(), nullable=True),
|
||||
sa.Column("verified_at", sa.DateTime(timezone=True), nullable=True),
|
||||
sa.Column("updated_at", sa.DateTime(timezone=True), nullable=True),
|
||||
)
|
||||
|
||||
|
||||
def downgrade() -> None:
|
||||
op.drop_table("exif_projections")
|
||||
@@ -1,4 +1,4 @@
|
||||
"""Application management CLI: ``python -m photo_pipeline {serve,migrate,worker,import-legacy-scores}``."""
|
||||
"""Application management CLI: ``python -m photo_pipeline {serve,migrate}``."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
@@ -16,17 +16,6 @@ def main(argv: Sequence[str] | None = None) -> int:
|
||||
commands.add_parser("migrate", help="Upgrade the database to the latest revision")
|
||||
worker_cmd = commands.add_parser("worker", help="Run a durable-job worker")
|
||||
worker_cmd.add_argument("--id", default="worker-1", help="Worker id (lease owner)")
|
||||
import_cmd = commands.add_parser(
|
||||
"import-legacy-scores",
|
||||
help="Import the archived CLI's nsfw_scores.csv into the database (US07-01)",
|
||||
)
|
||||
import_cmd.add_argument("csv", help="Path to nsfw_scores.csv")
|
||||
import_cmd.add_argument(
|
||||
"--overwrite", action="store_true", help="Replace differing imported scores"
|
||||
)
|
||||
import_cmd.add_argument(
|
||||
"--dry-run", action="store_true", help="Report what would happen and change nothing"
|
||||
)
|
||||
args = parser.parse_args(argv)
|
||||
|
||||
config = Config.from_env()
|
||||
@@ -36,24 +25,6 @@ def main(argv: Sequence[str] | None = None) -> int:
|
||||
run_migrations(config.database_url)
|
||||
return 0
|
||||
|
||||
if args.command == "import-legacy-scores":
|
||||
import json
|
||||
|
||||
from photo_pipeline.db import create_db_engine, create_session_factory
|
||||
from photo_pipeline.services.legacy_import import LegacyImportService, write_report
|
||||
|
||||
run_migrations(config.database_url)
|
||||
engine = create_db_engine(config.database_url)
|
||||
service = LegacyImportService(create_session_factory(engine))
|
||||
report = service.import_nsfw_scores(
|
||||
args.csv, overwrite=args.overwrite, dry_run=args.dry_run
|
||||
)
|
||||
# The report is the point: an import nobody can audit is not a migration.
|
||||
if not args.dry_run:
|
||||
write_report(report, config.data_dir)
|
||||
print(json.dumps(report.counts, indent=2))
|
||||
return 0
|
||||
|
||||
if args.command == "worker":
|
||||
from photo_pipeline.db import create_db_engine, create_session_factory
|
||||
|
||||
|
||||
@@ -8,15 +8,11 @@ and exposes the versioned ``/api/v1`` surface; US01-02 ships only health.
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from contextlib import asynccontextmanager
|
||||
from pathlib import Path
|
||||
|
||||
from fastapi import FastAPI, Request
|
||||
from fastapi.exceptions import RequestValidationError
|
||||
from fastapi.responses import JSONResponse
|
||||
from fastapi import FastAPI
|
||||
from fastapi.staticfiles import StaticFiles
|
||||
from starlette.exceptions import HTTPException as StarletteHTTPException
|
||||
|
||||
from photo_pipeline.api.routes import (
|
||||
albums,
|
||||
@@ -29,58 +25,20 @@ from photo_pipeline.api.routes import (
|
||||
library,
|
||||
renames,
|
||||
safety,
|
||||
session as session_routes,
|
||||
thumbnails,
|
||||
uploads,
|
||||
workflow,
|
||||
)
|
||||
from photo_pipeline.api.security import DEFAULT_HEADERS, SecurityMiddleware, Session
|
||||
|
||||
# Registers the safety_score / analysis job handlers on import.
|
||||
import photo_pipeline.jobs.domain_handlers # noqa: F401
|
||||
from photo_pipeline.config import Config
|
||||
from photo_pipeline.db import create_db_engine, create_session_factory, run_migrations
|
||||
from photo_pipeline.logging import configure_logging
|
||||
from photo_pipeline.services.thumbnails import ThumbnailService
|
||||
from photo_pipeline.services.upload_batches import UploadBatchService
|
||||
|
||||
FRONTEND_DIR = Path(__file__).resolve().parents[2] / "frontend"
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def _envelope(status: int, code: str, message: str) -> JSONResponse:
|
||||
return JSONResponse(
|
||||
status_code=status,
|
||||
content={"error": {"code": code, "message": message}},
|
||||
headers=DEFAULT_HEADERS,
|
||||
)
|
||||
|
||||
|
||||
def _install_error_handlers(app: FastAPI) -> None:
|
||||
"""One JSON error envelope everywhere, and nothing behind it.
|
||||
|
||||
An unhandled exception carries the library's absolute paths, SQL, and sometimes
|
||||
a credential in its text; the client gets a code, the operator gets the traceback
|
||||
in the server log (US07-02).
|
||||
"""
|
||||
|
||||
@app.exception_handler(StarletteHTTPException)
|
||||
async def _http_error(request: Request, exc: StarletteHTTPException):
|
||||
return _envelope(exc.status_code, "http_error", str(exc.detail))
|
||||
|
||||
@app.exception_handler(RequestValidationError)
|
||||
async def _validation_error(request: Request, exc: RequestValidationError):
|
||||
# Field locations only: the echoed input can be the caller's own data, but it
|
||||
# is also what ends up in shared logs and screenshots.
|
||||
fields = sorted(".".join(str(part) for part in error["loc"]) for error in exc.errors())
|
||||
return _envelope(422, "invalid_request", f"invalid request fields: {', '.join(fields)}")
|
||||
|
||||
@app.exception_handler(Exception)
|
||||
async def _unhandled(request: Request, exc: Exception):
|
||||
log.exception("unhandled error serving %s", request.url.path)
|
||||
return _envelope(500, "internal_error", "internal error")
|
||||
|
||||
|
||||
def create_app(config: Config | None = None) -> FastAPI:
|
||||
config = config or Config.from_env()
|
||||
@@ -97,9 +55,6 @@ def create_app(config: Config | None = None) -> FastAPI:
|
||||
# An upload whose process died left no outcome behind; resolve it now so the
|
||||
# uploader lane is free and the uncertain batch is visible (US05-02).
|
||||
UploadBatchService(app.state.session_factory, config=config).recover()
|
||||
# A render killed mid-write leaves its temporary beside the cache entry;
|
||||
# remove those recognized leftovers, and only those (US07-03).
|
||||
ThumbnailService(app.state.session_factory, config).cleanup_temp_files()
|
||||
try:
|
||||
yield
|
||||
finally:
|
||||
@@ -107,12 +62,6 @@ def create_app(config: Config | None = None) -> FastAPI:
|
||||
app.state.engine = None
|
||||
|
||||
app = FastAPI(title="Photo Pipeline", version="0.1.0", lifespan=lifespan)
|
||||
# One session per process: the browser exchanges it for a cookie + CSRF token,
|
||||
# and every other origin is refused before a route ever runs (US07-02).
|
||||
app.state.session = Session.create()
|
||||
app.add_middleware(SecurityMiddleware, session=app.state.session, config=config)
|
||||
_install_error_handlers(app)
|
||||
app.include_router(session_routes.router, prefix="/api/v1")
|
||||
app.include_router(health.router, prefix="/api/v1")
|
||||
app.include_router(inventory.router, prefix="/api/v1")
|
||||
app.include_router(duplicates.router, prefix="/api/v1")
|
||||
|
||||
@@ -18,10 +18,7 @@ router = APIRouter(tags=["analysis"])
|
||||
|
||||
|
||||
def _service(request: Request) -> AnalysisService:
|
||||
return AnalysisService(
|
||||
request.app.state.session_factory,
|
||||
library_roots=tuple(request.app.state.config.library_roots),
|
||||
)
|
||||
return AnalysisService(request.app.state.session_factory)
|
||||
|
||||
|
||||
def _error(status: int, code: str, message: str) -> JSONResponse:
|
||||
|
||||
@@ -1,30 +0,0 @@
|
||||
"""Session bootstrap: the one endpoint reachable without a session (US07-02).
|
||||
|
||||
It sets the ``HttpOnly``/``SameSite=Strict`` session cookie and returns the CSRF
|
||||
token in the body. A foreign page can call this — it just cannot read the answer,
|
||||
because the app sends no CORS headers — and the cookie it received is never attached
|
||||
to a request that foreign page initiates.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from fastapi import APIRouter, Request
|
||||
from fastapi.responses import JSONResponse
|
||||
|
||||
from photo_pipeline.api.security import SESSION_COOKIE
|
||||
|
||||
router = APIRouter(tags=["session"])
|
||||
|
||||
|
||||
@router.get("/session")
|
||||
def start_session(request: Request) -> JSONResponse:
|
||||
session = request.app.state.session
|
||||
response = JSONResponse({"csrf_token": session.csrf_token})
|
||||
response.set_cookie(
|
||||
SESSION_COOKIE,
|
||||
session.id,
|
||||
httponly=True,
|
||||
samesite="strict",
|
||||
path="/",
|
||||
)
|
||||
return response
|
||||
@@ -31,7 +31,5 @@ def get_thumbnail(asset_id: str, request: Request, size: int = Query(512)):
|
||||
return FileResponse(
|
||||
path,
|
||||
media_type="image/webp",
|
||||
# private: the URL is versioned and immutable, but these bytes are the user's
|
||||
# photos and must never sit in a shared cache (US07-02).
|
||||
headers={"Cache-Control": "private, max-age=31536000, immutable"},
|
||||
headers={"Cache-Control": "public, max-age=31536000, immutable"},
|
||||
)
|
||||
|
||||
@@ -1,192 +0,0 @@
|
||||
"""Local-web attack surface: session, CSRF, Origin/Host checks, default headers.
|
||||
|
||||
The app binds to 127.0.0.1, so the attacker is not a remote client but another page
|
||||
in the user's browser (concept §15, "Local web attack"): any site can issue requests
|
||||
to ``http://127.0.0.1:8000`` and can embed ``<img src=...>`` against media endpoints.
|
||||
The defenses stack, because each one alone has a hole:
|
||||
|
||||
* **Host** must be a loopback name — a DNS rebinding host that resolves to 127.0.0.1
|
||||
passes the browser's origin rules but not this check.
|
||||
* **Origin**, when the browser sends one, must be this exact origin (scheme, host,
|
||||
port). There is no CORS middleware at all, so a foreign page can never *read* a
|
||||
response even if it manages to send a request.
|
||||
* **Sec-Fetch-Site** rejects cross-site loads that carry no Origin, which is what an
|
||||
``<img>`` or ``<script>`` against a media endpoint looks like.
|
||||
* A **session cookie** (``SameSite=Strict``, ``HttpOnly``) is required by every
|
||||
``/api/v1`` route except liveness/readiness and the bootstrap itself. Strict means
|
||||
the browser never attaches it to a request another site initiated.
|
||||
* A **CSRF token** must be echoed in a header on every mutation. It is handed out
|
||||
only in the bootstrap response body, which a foreign page cannot read (no CORS) —
|
||||
so possessing it proves the caller is same-origin.
|
||||
|
||||
``evaluate`` is a pure function over the request metadata: the whole policy is one
|
||||
table that a unit test can enumerate, and the middleware only applies its verdict.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import secrets
|
||||
from collections.abc import Mapping
|
||||
from dataclasses import dataclass
|
||||
from urllib.parse import urlsplit
|
||||
|
||||
from starlette.datastructures import Headers, MutableHeaders
|
||||
from starlette.responses import JSONResponse
|
||||
|
||||
SESSION_COOKIE = "pp_session"
|
||||
CSRF_HEADER = "x-csrf-token"
|
||||
API_PREFIX = "/api/v1"
|
||||
SAFE_METHODS = frozenset({"GET", "HEAD", "OPTIONS"})
|
||||
# Reachable without a session: liveness/readiness (an orchestrator has no cookie)
|
||||
# and the bootstrap that issues the session in the first place.
|
||||
PUBLIC_PATHS = frozenset(
|
||||
{f"{API_PREFIX}/health/live", f"{API_PREFIX}/health/ready", f"{API_PREFIX}/session"}
|
||||
)
|
||||
LOOPBACK_HOSTS = frozenset({"127.0.0.1", "localhost", "::1", "[::1]"})
|
||||
|
||||
# Applied to every response. No inline script/style is used by the frontend, so the
|
||||
# policy can stay strict; `frame-ancestors 'none'` and CORP keep other pages from
|
||||
# embedding the app or its thumbnails.
|
||||
DEFAULT_HEADERS = {
|
||||
"x-content-type-options": "nosniff",
|
||||
"x-frame-options": "DENY",
|
||||
"referrer-policy": "no-referrer",
|
||||
"cross-origin-resource-policy": "same-origin",
|
||||
"cross-origin-opener-policy": "same-origin",
|
||||
"content-security-policy": (
|
||||
"default-src 'self'; img-src 'self' data:; style-src 'self'; script-src 'self'; "
|
||||
"connect-src 'self'; frame-ancestors 'none'; base-uri 'none'; form-action 'none'"
|
||||
),
|
||||
}
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Session:
|
||||
"""One process, one session. A local app has exactly one user; a session store
|
||||
would be bookkeeping without a second subject to distinguish.
|
||||
|
||||
ponytail: per-session rows if the app ever serves more than one operator.
|
||||
"""
|
||||
|
||||
id: str
|
||||
csrf_token: str
|
||||
|
||||
@classmethod
|
||||
def create(cls) -> Session:
|
||||
return cls(secrets.token_urlsafe(32), secrets.token_urlsafe(32))
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Refusal:
|
||||
status: int
|
||||
code: str
|
||||
message: str
|
||||
|
||||
|
||||
def split_host(value: str) -> tuple[str, str]:
|
||||
"""``"127.0.0.1:8000"`` -> ``("127.0.0.1", "8000")``; bracketed IPv6 aware."""
|
||||
value = value.strip()
|
||||
if value.startswith("["):
|
||||
host, _, port = value.partition("]")
|
||||
return host + "]", port.lstrip(":")
|
||||
host, _, port = value.partition(":")
|
||||
return host, port
|
||||
|
||||
|
||||
def evaluate(
|
||||
*,
|
||||
method: str,
|
||||
path: str,
|
||||
headers: Mapping[str, str],
|
||||
session: Session,
|
||||
allowed_hosts: frozenset[str] = LOOPBACK_HOSTS,
|
||||
max_request_bytes: int,
|
||||
) -> Refusal | None:
|
||||
"""Why this request must be refused, or ``None`` when it may proceed."""
|
||||
host_header = headers.get("host", "")
|
||||
host, port = split_host(host_header)
|
||||
if host.lower() not in allowed_hosts:
|
||||
return Refusal(403, "host_not_allowed", "request host is not a local address")
|
||||
|
||||
origin = headers.get("origin")
|
||||
if origin is not None and origin != "":
|
||||
parts = urlsplit(origin)
|
||||
origin_host, origin_port = split_host(parts.netloc)
|
||||
if (
|
||||
parts.scheme not in ("http", "https")
|
||||
or origin_host.lower() not in allowed_hosts
|
||||
or origin_port != port
|
||||
):
|
||||
return Refusal(403, "origin_not_allowed", "request origin is not this application")
|
||||
|
||||
# Absent means a non-browser client; "none" is a user-initiated navigation.
|
||||
fetch_site = headers.get("sec-fetch-site")
|
||||
if fetch_site is not None and fetch_site not in ("same-origin", "none"):
|
||||
return Refusal(403, "cross_site_blocked", "cross-site requests are not accepted")
|
||||
|
||||
length = headers.get("content-length")
|
||||
if length and length.isdigit() and int(length) > max_request_bytes:
|
||||
return Refusal(413, "payload_too_large", "request body exceeds the configured limit")
|
||||
|
||||
protected = path.startswith(API_PREFIX) and path not in PUBLIC_PATHS
|
||||
if not protected:
|
||||
return None
|
||||
|
||||
if headers.get("cookie-session") != session.id:
|
||||
return Refusal(401, "unauthenticated", "a valid application session is required")
|
||||
if method.upper() not in SAFE_METHODS and headers.get(CSRF_HEADER) != session.csrf_token:
|
||||
return Refusal(403, "csrf_failed", "missing or invalid CSRF token")
|
||||
return None
|
||||
|
||||
|
||||
class SecurityMiddleware:
|
||||
"""Pure-ASGI so the SSE stream keeps streaming (BaseHTTPMiddleware buffers)."""
|
||||
|
||||
def __init__(self, app, *, session: Session, config) -> None:
|
||||
self.app = app
|
||||
self.session = session
|
||||
self.max_request_bytes = config.max_request_bytes
|
||||
self.allowed_hosts = frozenset(LOOPBACK_HOSTS | {str(config.host).lower()})
|
||||
|
||||
async def __call__(self, scope, receive, send) -> None:
|
||||
if scope["type"] != "http":
|
||||
await self.app(scope, receive, send)
|
||||
return
|
||||
headers = Headers(scope=scope)
|
||||
# The cookie is read here and handed to the pure policy as one value, so the
|
||||
# policy never has to parse a Cookie header.
|
||||
lookup = dict(headers)
|
||||
lookup["cookie-session"] = _cookie(headers.get("cookie", ""), SESSION_COOKIE)
|
||||
refusal = evaluate(
|
||||
method=scope.get("method", "GET"),
|
||||
path=scope.get("path", "/"),
|
||||
headers=lookup,
|
||||
session=self.session,
|
||||
allowed_hosts=self.allowed_hosts,
|
||||
max_request_bytes=self.max_request_bytes,
|
||||
)
|
||||
if refusal is not None:
|
||||
response = JSONResponse(
|
||||
status_code=refusal.status,
|
||||
content={"error": {"code": refusal.code, "message": refusal.message}},
|
||||
headers=DEFAULT_HEADERS,
|
||||
)
|
||||
await response(scope, receive, send)
|
||||
return
|
||||
|
||||
async def send_with_headers(message):
|
||||
if message["type"] == "http.response.start":
|
||||
out = MutableHeaders(scope=message)
|
||||
for name, value in DEFAULT_HEADERS.items():
|
||||
out.setdefault(name, value)
|
||||
await send(message)
|
||||
|
||||
await self.app(scope, receive, send_with_headers)
|
||||
|
||||
|
||||
def _cookie(header: str, name: str) -> str:
|
||||
for part in header.split(";"):
|
||||
key, _, value = part.strip().partition("=")
|
||||
if key == name:
|
||||
return value
|
||||
return ""
|
||||
@@ -30,10 +30,6 @@ class Config(BaseModel):
|
||||
log_level: str = "INFO"
|
||||
log_format: str = "json" # "json" or "text"
|
||||
|
||||
# Largest request body the API accepts. Every endpoint takes small JSON commands;
|
||||
# anything larger is a mistake or an attempt to exhaust memory (US07-02).
|
||||
max_request_bytes: int = 1_048_576
|
||||
|
||||
# Library boundary for path validation (os.pathsep-separated in the env var).
|
||||
library_roots: tuple[Path, ...] = ()
|
||||
thumbnail_cache_quota_bytes: int = 500_000_000
|
||||
|
||||
@@ -1,119 +0,0 @@
|
||||
"""Bounded, defensive image decoding — one door for every pixel this app reads.
|
||||
|
||||
A photo library contains files nobody planned for: truncated downloads, zero-byte
|
||||
placeholders, a PNG whose header claims 200000×200000, a TIFF with a broken ICC
|
||||
profile, an extension that lies about its content. None of them may take down a
|
||||
request or a worker, and none may decode more pixels than the caller allowed
|
||||
(concept §17: decoded pixels, not file size, are what exhausts memory).
|
||||
|
||||
``open_image`` is that single door:
|
||||
|
||||
* the declared dimensions are checked **before** a pixel is decoded;
|
||||
* Pillow's decompression-bomb *warning* is promoted to an error, because the
|
||||
warning band (between Pillow's limit and twice it) still decodes the image;
|
||||
* every decoder failure — at open time or during the caller's decode — becomes one
|
||||
of two typed errors, so callers map them to their own item state instead of
|
||||
catching ``Exception``;
|
||||
* error text names no path: it reaches API responses, and the full reason goes to
|
||||
the server log instead (US07-02).
|
||||
|
||||
``to_srgb`` and ``draft`` are the other two bounded-decode helpers: colour-manage a
|
||||
profile-bearing image into sRGB, and let JPEG decode straight to a size near the
|
||||
requested one rather than at full resolution.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import io
|
||||
import logging
|
||||
import warnings
|
||||
from contextlib import contextmanager
|
||||
from pathlib import Path
|
||||
|
||||
from PIL import Image, ImageCms, ImageFile, UnidentifiedImageError
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
# Matches ``Config.thumbnail_max_pixels``; used where no configuration is at hand
|
||||
# (hashing runs inside discovery, which takes no config).
|
||||
DEFAULT_MAX_PIXELS = 100_000_000
|
||||
|
||||
|
||||
class MediaError(Exception):
|
||||
"""A file could not be turned into pixels safely."""
|
||||
|
||||
|
||||
class UndecodableImage(MediaError):
|
||||
"""Corrupt, truncated, empty, or not an image at all."""
|
||||
|
||||
|
||||
class ImageTooLarge(MediaError):
|
||||
"""More pixels than this operation is allowed to decode."""
|
||||
|
||||
|
||||
@contextmanager
|
||||
def open_image(path: Path | str, *, max_pixels: int = DEFAULT_MAX_PIXELS):
|
||||
"""Yield an open :class:`PIL.Image.Image`, bounded and with typed failures.
|
||||
|
||||
Decoder errors raised inside the ``with`` body are translated too — a truncated
|
||||
JPEG only fails when its pixels are actually pulled, which is the caller's line,
|
||||
not this one.
|
||||
"""
|
||||
# Pillow's truncation tolerance is a process-global switch that any library in
|
||||
# the process can flip (the donor CLI did). This door decides the policy for its
|
||||
# own callers: half a file is not a picture.
|
||||
tolerated = ImageFile.LOAD_TRUNCATED_IMAGES
|
||||
ImageFile.LOAD_TRUNCATED_IMAGES = False
|
||||
with warnings.catch_warnings():
|
||||
# The warning band is not a warning for us: it means Pillow was willing to
|
||||
# decode an image large enough to be a denial-of-service.
|
||||
warnings.simplefilter("error", Image.DecompressionBombWarning)
|
||||
try:
|
||||
with Image.open(path) as image:
|
||||
width, height = image.size
|
||||
if width * height > max_pixels:
|
||||
raise ImageTooLarge(f"{width}x{height} exceeds the {max_pixels} pixel limit")
|
||||
yield image
|
||||
except MediaError:
|
||||
raise
|
||||
except (Image.DecompressionBombError, Image.DecompressionBombWarning) as error:
|
||||
log.info("refused oversized image %s: %s", path, error)
|
||||
raise ImageTooLarge("image exceeds the decompression-bomb limit") from error
|
||||
except (UnidentifiedImageError, OSError, ValueError, SyntaxError, MemoryError) as error:
|
||||
log.info("cannot decode %s: %s", path, error)
|
||||
raise UndecodableImage(f"cannot decode image ({type(error).__name__})") from error
|
||||
finally:
|
||||
ImageFile.LOAD_TRUNCATED_IMAGES = tolerated
|
||||
|
||||
|
||||
def draft(image: Image.Image, size: int) -> None:
|
||||
"""Ask the decoder for a smaller image where the format allows it (JPEG).
|
||||
|
||||
This is the difference between decoding a 40-megapixel JPEG and decoding the
|
||||
roughly 1-megapixel version a 1280px preview needs.
|
||||
"""
|
||||
try:
|
||||
image.draft(None, (size, size))
|
||||
except (AttributeError, ValueError, OSError): # not a draft-capable format
|
||||
pass
|
||||
|
||||
|
||||
def to_srgb(image: Image.Image, *, mode: str) -> Image.Image:
|
||||
"""Convert into ``mode``, colour-managing through an embedded ICC profile.
|
||||
|
||||
Without this a wide-gamut original renders with visibly wrong colours, because
|
||||
its numbers are interpreted as sRGB. A broken or unreadable profile is not a
|
||||
reason to fail a preview — the plain conversion is still a correct picture.
|
||||
"""
|
||||
profile = image.info.get("icc_profile")
|
||||
if profile:
|
||||
try:
|
||||
return ImageCms.profileToProfile(
|
||||
image,
|
||||
ImageCms.ImageCmsProfile(io.BytesIO(profile)),
|
||||
ImageCms.createProfile("sRGB"),
|
||||
outputMode=mode,
|
||||
)
|
||||
except Exception as error: # noqa: BLE001 - any ICC failure falls back
|
||||
log.info("ignoring unusable ICC profile on %s: %s", getattr(image, "filename", "?"), error)
|
||||
return image.convert(mode)
|
||||
@@ -56,30 +56,6 @@ def read_keyword_sets(paths: Iterable[str]) -> dict[str, set[str]]:
|
||||
return out
|
||||
|
||||
|
||||
def read_all(path: str) -> dict | None:
|
||||
"""Every tag exiftool can read from ``path``, or ``None`` when it cannot answer.
|
||||
|
||||
This is the snapshot an EXIF checkpoint compares against: proving that a write
|
||||
preserved the fields it does not own requires knowing all of them, not just the
|
||||
ones being written (US07-03). ``None`` (exiftool missing, unreadable file,
|
||||
unparsable output) is not an empty snapshot — a caller must not read it as
|
||||
"nothing was there".
|
||||
"""
|
||||
try:
|
||||
result = subprocess.run(
|
||||
["exiftool", "-m", "-j", "-G0:1", path], capture_output=True, text=True
|
||||
)
|
||||
except FileNotFoundError:
|
||||
return None
|
||||
try:
|
||||
records = json.loads(result.stdout or "[]")
|
||||
except ValueError:
|
||||
return None
|
||||
if not records:
|
||||
return None
|
||||
return {k: v for k, v in records[0].items() if k != "SourceFile"}
|
||||
|
||||
|
||||
def apply_keywords(path: str, *, add: Iterable[str] = (), remove: Iterable[str] = ()) -> bool:
|
||||
"""Idempotently add/remove keywords in Keywords + Subject; preserve all else."""
|
||||
args = ["exiftool", "-m", "-overwrite_original"]
|
||||
|
||||
@@ -15,8 +15,6 @@ from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
from photo_pipeline import imaging
|
||||
|
||||
MODEL_ID = "AdamCodd/vit-base-nsfw-detector"
|
||||
BATCH = 16
|
||||
|
||||
@@ -56,17 +54,13 @@ class NsfwModel:
|
||||
self._ensure_loaded()
|
||||
import numpy as np
|
||||
import torch
|
||||
from PIL import Image
|
||||
from PIL import Image, ImageFile
|
||||
|
||||
# The donor set ``ImageFile.LOAD_TRUNCATED_IMAGES = True`` here. That flag is
|
||||
# process-global: in this application the same process also hashes files and
|
||||
# renders previews, and those must keep failing loudly on a truncated file
|
||||
# rather than quietly working on half of one (US07-03). An unreadable image
|
||||
# is skipped instead — it stays unscored, and therefore visibly undecided.
|
||||
def preprocess(path):
|
||||
with imaging.open_image(path) as image:
|
||||
small = image.convert("RGB").resize((self._size, self._size), Image.BILINEAR)
|
||||
array = (np.asarray(small, dtype="float32") / 255.0 - 0.5) / 0.5
|
||||
ImageFile.LOAD_TRUNCATED_IMAGES = True
|
||||
|
||||
def preprocess(image):
|
||||
image = image.convert("RGB").resize((self._size, self._size), Image.BILINEAR)
|
||||
array = (np.asarray(image, dtype="float32") / 255.0 - 0.5) / 0.5
|
||||
return torch.from_numpy(array).permute(2, 0, 1)
|
||||
|
||||
results: list[tuple[str, float]] = []
|
||||
@@ -75,10 +69,9 @@ class NsfwModel:
|
||||
tensors, batch_paths = [], []
|
||||
for path in items[start : start + self.batch]:
|
||||
try:
|
||||
tensors.append(preprocess(path))
|
||||
tensors.append(preprocess(Image.open(path)))
|
||||
batch_paths.append(path)
|
||||
except (imaging.MediaError, OSError, ValueError):
|
||||
# One bad file must not cost the batch its other fifteen.
|
||||
except Exception:
|
||||
continue
|
||||
if not tensors:
|
||||
continue
|
||||
|
||||
@@ -38,8 +38,7 @@ def _safety_score_item(asset_id: str, ctx: JobContext) -> None:
|
||||
def _analysis_item(asset_id: str, ctx: JobContext) -> None:
|
||||
from photo_pipeline.services.analysis import AnalysisService
|
||||
|
||||
roots = tuple(getattr(ctx.config, "library_roots", ()) or ())
|
||||
AnalysisService(ctx.session_factory, library_roots=roots).run([asset_id])
|
||||
AnalysisService(ctx.session_factory).run([asset_id])
|
||||
|
||||
|
||||
def _upload_batch_item(batch_id: str, ctx: JobContext) -> None:
|
||||
|
||||
@@ -7,7 +7,6 @@ Alembic environment relies on.
|
||||
from photo_pipeline.models.albums import AlbumProposal
|
||||
from photo_pipeline.models.archives import ArchiveLocation, ArchiveOperation, ArchivePlan
|
||||
from photo_pipeline.models.assets import Asset, AssetPath
|
||||
from photo_pipeline.models.exif import ExifProjection
|
||||
from photo_pipeline.models.duplicates import (
|
||||
DuplicateCluster,
|
||||
DuplicateMember,
|
||||
@@ -29,7 +28,6 @@ __all__ = [
|
||||
"DuplicateCluster",
|
||||
"DuplicateMember",
|
||||
"DuplicateNegativeLink",
|
||||
"ExifProjection",
|
||||
"Job",
|
||||
"JobItem",
|
||||
"JobEvent",
|
||||
|
||||
@@ -1,33 +0,0 @@
|
||||
"""The durable EXIF projection per asset and stage (concept §3, US07-03).
|
||||
|
||||
One row per ``(asset_id, stage)``: what the stage wanted written, what the file
|
||||
looked like afterwards, and whether anything outside the stage's ownership moved.
|
||||
``state = divergent`` is the whole point of the table — it survives restarts, keeps
|
||||
the asset out of stages that require verified metadata, and gives a human something
|
||||
to look at instead of a silent repair.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime
|
||||
|
||||
from sqlalchemy import DateTime, ForeignKey, Integer, String
|
||||
from sqlalchemy.orm import Mapped, mapped_column
|
||||
|
||||
from photo_pipeline.db import Base
|
||||
|
||||
|
||||
class ExifProjection(Base):
|
||||
__tablename__ = "exif_projections"
|
||||
|
||||
asset_id: Mapped[str] = mapped_column(ForeignKey("assets.id"), primary_key=True)
|
||||
stage: Mapped[str] = mapped_column(String, primary_key=True) # safety | analysis
|
||||
id: Mapped[str] = mapped_column(String, nullable=False)
|
||||
projection_version: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
|
||||
desired_json: Mapped[str | None] = mapped_column(String)
|
||||
divergent_fields: Mapped[str | None] = mapped_column(String) # JSON array
|
||||
result_file_sha256: Mapped[str | None] = mapped_column(String)
|
||||
state: Mapped[str] = mapped_column(String, nullable=False) # verified|divergent|failed
|
||||
error_code: Mapped[str | None] = mapped_column(String)
|
||||
verified_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
|
||||
updated_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
|
||||
@@ -49,29 +49,6 @@ def resolve_within(root: Path, path: os.PathLike | str) -> Path:
|
||||
return resolved
|
||||
|
||||
|
||||
def resolve_in_roots(roots: Iterable[os.PathLike | str], path: os.PathLike | str) -> Path:
|
||||
"""The resolved path, proven to be inside one of ``roots`` and not excluded.
|
||||
|
||||
Callers must use the **returned** path for whatever they do next: validating one
|
||||
name and then opening another is the symlink race this exists to close (US07-02).
|
||||
The message names no path — it reaches API responses.
|
||||
|
||||
With no roots configured there is no boundary to check; that is a property of the
|
||||
configuration, not permission granted to this call.
|
||||
"""
|
||||
if is_excluded(path):
|
||||
raise PathPolicyError("path is inside an excluded (_IGNORE/) tree")
|
||||
roots = list(roots)
|
||||
if not roots:
|
||||
return Path(path)
|
||||
for root in roots:
|
||||
try:
|
||||
return resolve_within(Path(root), path)
|
||||
except PathPolicyError:
|
||||
continue
|
||||
raise PathPolicyError("path is outside the configured library roots")
|
||||
|
||||
|
||||
def iter_supported_files(root: os.PathLike | str) -> Iterator[Path]:
|
||||
"""Yield supported, non-excluded files under ``root`` in deterministic order.
|
||||
|
||||
|
||||
@@ -25,9 +25,9 @@ from typing import Protocol
|
||||
from sqlalchemy import func, select
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
|
||||
from photo_pipeline import path_policy
|
||||
from photo_pipeline.integrations import exiftool
|
||||
from photo_pipeline.models import AnalysisResult, Asset, SafetyReview
|
||||
from photo_pipeline.services import exif_checkpoint
|
||||
from photo_pipeline.services import hashing
|
||||
from photo_pipeline.services.safety import SFW
|
||||
|
||||
MODEL = "gemini-2.5-flash"
|
||||
@@ -59,16 +59,9 @@ def _now() -> datetime:
|
||||
|
||||
|
||||
class AnalysisService:
|
||||
def __init__(
|
||||
self,
|
||||
session_factory: sessionmaker,
|
||||
*,
|
||||
provider: VisionProvider | None = None,
|
||||
library_roots: tuple = (),
|
||||
) -> None:
|
||||
def __init__(self, session_factory: sessionmaker, *, provider: VisionProvider | None = None) -> None:
|
||||
self._session_factory = session_factory
|
||||
self._provider = provider
|
||||
self._roots = tuple(library_roots)
|
||||
|
||||
def _sfw_asset_ids(self, session) -> set[str]:
|
||||
"""Asset ids whose latest safety decision is ``sfw`` — the ONLY assets that
|
||||
@@ -143,18 +136,6 @@ class AnalysisService:
|
||||
if not path:
|
||||
skipped += 1
|
||||
continue
|
||||
# Second gate, at the moment of use: the database says where the file
|
||||
# was, the filesystem decides what that name means now. A link swapped
|
||||
# under an asset after the scan would otherwise send bytes from outside
|
||||
# the library — the one place that leaves this machine (US07-02).
|
||||
try:
|
||||
path = str(path_policy.resolve_in_roots(self._roots, path))
|
||||
except path_policy.PathPolicyError as error:
|
||||
self._store(
|
||||
asset_id, status="error", result=None, error=str(error), tokens=0, raw=""
|
||||
)
|
||||
errors += 1
|
||||
continue
|
||||
try:
|
||||
result = provider.analyze(path, album_hint=_album_hint(path))
|
||||
except Exception as error: # provider/validation failure is per-asset
|
||||
@@ -169,7 +150,7 @@ class AnalysisService:
|
||||
tokens=int(result.get("_tokens", 0)) if isinstance(result, dict) else 0,
|
||||
raw=json.dumps(result, ensure_ascii=False),
|
||||
)
|
||||
self._write_analysis_exif(asset_id, path, result)
|
||||
_write_analysis_exif(path, result)
|
||||
analyzed += 1
|
||||
return {"analyzed": analyzed, "skipped": skipped, "errors": errors}
|
||||
|
||||
@@ -194,42 +175,10 @@ class AnalysisService:
|
||||
row.location_hint = result.get("location_hint")
|
||||
row.approx_year = result.get("approx_year")
|
||||
row.analyzed_at = now
|
||||
row.exif_written_at = now
|
||||
session.add(row)
|
||||
session.commit()
|
||||
|
||||
def _write_analysis_exif(self, asset_id: str, path: str, result: dict) -> None:
|
||||
"""The analysis EXIF checkpoint: additive keywords, then prove the rest held.
|
||||
|
||||
Additive by design — safety keywords and the user's own keywords are merged
|
||||
with, never replaced (concept §3). ``exif_written_at`` is set only when the
|
||||
read-back verified both the new keywords and every field this stage does not
|
||||
own; a divergent result is recorded and left for a human (US07-03).
|
||||
ponytail: the managed ``AI:`` caption segment, once captions are owned here.
|
||||
"""
|
||||
tags = tuple(str(tag) for tag in (result.get("tags") or []))
|
||||
if not tags:
|
||||
return
|
||||
checkpoint = exif_checkpoint.run(path, add=tags)
|
||||
exif_checkpoint.record(
|
||||
self._session_factory,
|
||||
asset_id=asset_id,
|
||||
stage="analysis",
|
||||
result=checkpoint,
|
||||
add=tags,
|
||||
)
|
||||
if not checkpoint.verified:
|
||||
return
|
||||
with self._session_factory() as session:
|
||||
row = session.get(AnalysisResult, asset_id)
|
||||
if row is not None:
|
||||
row.exif_written_at = checkpoint.verified_at
|
||||
asset = session.get(Asset, asset_id)
|
||||
if asset is not None and checkpoint.sha256:
|
||||
# The bytes changed when the container was rewritten; upload must use
|
||||
# the hash of what is actually on disk now (concept §3).
|
||||
asset.current_sha256 = checkpoint.sha256
|
||||
session.commit()
|
||||
|
||||
def get(self, asset_id: str) -> dict | None:
|
||||
with self._session_factory() as session:
|
||||
row = session.get(AnalysisResult, asset_id)
|
||||
@@ -242,6 +191,17 @@ def _album_hint(path: str) -> str:
|
||||
return Path(path).parent.name
|
||||
|
||||
|
||||
def _write_analysis_exif(path: str, result: dict) -> None:
|
||||
"""Additive analysis keywords into EXIF (Keywords/Subject), preserving safety
|
||||
and user keywords. The donor also wrote a managed caption; only keywords are
|
||||
written here via the shared adapter.
|
||||
ponytail: add the managed ``AI:`` caption segment + read-back preservation check
|
||||
when the analysis EXIF checkpoint is hardened."""
|
||||
tags = result.get("tags") if isinstance(result, dict) else None
|
||||
if tags:
|
||||
exiftool.apply_keywords(path, add=[str(t) for t in tags])
|
||||
|
||||
|
||||
def _result_dict(row: AnalysisResult) -> dict:
|
||||
data = {field: getattr(row, field) for field in RESULT_FIELDS}
|
||||
data["tags"] = json.loads(row.tags) if row.tags else []
|
||||
|
||||
@@ -1,173 +0,0 @@
|
||||
"""The EXIF checkpoint every metadata stage ends with (concept §3, US07-03).
|
||||
|
||||
A stage does not own a file's metadata; it owns a few fields in it. So writing is
|
||||
never "set these tags" — it is:
|
||||
|
||||
snapshot everything → write only the owned fields → read everything back
|
||||
→ prove the owned fields landed → prove nothing else moved
|
||||
→ refresh the file hash → record the projection
|
||||
|
||||
Non-destructive here means *semantic* preservation: exiftool may rewrite the whole
|
||||
container, so the file's bytes, size, and timestamps legitimately change. What may
|
||||
not change is any field this stage does not own. When one does, the checkpoint is
|
||||
``divergent``: the result is recorded, the stage is **not** marked verified, and
|
||||
nothing is silently repaired — a later stage that needs verified metadata (upload)
|
||||
therefore stays blocked until a human looks.
|
||||
|
||||
``failed`` is the third outcome and is deliberately distinct: exiftool missing, an
|
||||
unreadable file, or a write that did not take is not evidence that metadata is fine.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import uuid
|
||||
from dataclasses import dataclass
|
||||
from datetime import datetime, timezone
|
||||
|
||||
from photo_pipeline.integrations import exiftool
|
||||
from photo_pipeline.models import ExifProjection
|
||||
from photo_pipeline.services import hashing
|
||||
|
||||
# The tags the safety and analysis stages may change. Matched on the tag name, so
|
||||
# every group carries the same rule (IPTC:Keywords, XMP:XMP-dc:Subject, ...).
|
||||
OWNED_TAGS = frozenset({"Keywords", "Subject"})
|
||||
|
||||
# Not metadata about the picture: filesystem facts, the digest that necessarily
|
||||
# moves whenever IPTC does, and the structural tags exiftool has to create the first
|
||||
# time it writes an IPTC or XMP block. Comparing these would report every write as
|
||||
# divergent and make the signal worthless.
|
||||
VOLATILE_PREFIXES = ("File:System:", "ExifTool:")
|
||||
VOLATILE_KEYS = frozenset(
|
||||
{
|
||||
"File:CurrentIPTCDigest",
|
||||
"IPTC:ApplicationRecordVersion",
|
||||
"XMP:XMP-x:XMPToolkit",
|
||||
"XMP:XMP-xmp:MetadataDate",
|
||||
}
|
||||
)
|
||||
|
||||
VERIFIED = "verified"
|
||||
DIVERGENT = "divergent"
|
||||
FAILED = "failed"
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class CheckpointResult:
|
||||
state: str # verified | divergent | failed
|
||||
changed_fields: tuple[str, ...] = ()
|
||||
sha256: str | None = None
|
||||
verified_at: datetime | None = None
|
||||
reason: str | None = None
|
||||
|
||||
@property
|
||||
def verified(self) -> bool:
|
||||
return self.state == VERIFIED
|
||||
|
||||
|
||||
def _now() -> datetime:
|
||||
return datetime.now(timezone.utc)
|
||||
|
||||
|
||||
def is_owned(key: str) -> bool:
|
||||
return key.rsplit(":", 1)[-1] in OWNED_TAGS
|
||||
|
||||
|
||||
def is_volatile(key: str) -> bool:
|
||||
return key in VOLATILE_KEYS or key.startswith(VOLATILE_PREFIXES)
|
||||
|
||||
|
||||
def compare(before: dict, after: dict) -> tuple[str, ...]:
|
||||
"""Fields outside this stage's ownership whose value did not survive the write.
|
||||
|
||||
Additions count: a tag that appears out of nowhere is as much a divergence as a
|
||||
tag that disappeared — both mean the write did more than it was asked to.
|
||||
"""
|
||||
keys = set(before) | set(after)
|
||||
return tuple(
|
||||
sorted(
|
||||
key
|
||||
for key in keys
|
||||
if not is_owned(key)
|
||||
and not is_volatile(key)
|
||||
and before.get(key) != after.get(key)
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
def owned_values(snapshot: dict) -> set[str]:
|
||||
"""Lowercased Keywords/Subject values across every group in the snapshot."""
|
||||
values: set[str] = set()
|
||||
for key, value in snapshot.items():
|
||||
if not is_owned(key):
|
||||
continue
|
||||
items = value if isinstance(value, list) else [value]
|
||||
values.update(str(item).strip().lower() for item in items if item is not None)
|
||||
return values
|
||||
|
||||
|
||||
def run(
|
||||
path: str,
|
||||
*,
|
||||
add: tuple[str, ...] = (),
|
||||
remove: tuple[str, ...] = (),
|
||||
) -> CheckpointResult:
|
||||
"""Write the owned keywords for one asset and verify the whole file around them."""
|
||||
before = exiftool.read_all(path)
|
||||
if before is None:
|
||||
return CheckpointResult(FAILED, reason="metadata_unreadable")
|
||||
|
||||
if not exiftool.apply_keywords(path, add=add, remove=remove):
|
||||
return CheckpointResult(FAILED, reason="write_failed")
|
||||
|
||||
after = exiftool.read_all(path)
|
||||
if after is None:
|
||||
return CheckpointResult(FAILED, reason="readback_unreadable")
|
||||
|
||||
present = owned_values(after)
|
||||
wanted = {value.strip().lower() for value in add}
|
||||
unwanted = {value.strip().lower() for value in remove}
|
||||
if not wanted <= present or (unwanted & present):
|
||||
return CheckpointResult(FAILED, reason="owned_fields_not_written")
|
||||
|
||||
changed = compare(before, after)
|
||||
sha256 = hashing.sha256_file(path)
|
||||
if changed:
|
||||
return CheckpointResult(DIVERGENT, changed_fields=changed, sha256=sha256)
|
||||
return CheckpointResult(VERIFIED, sha256=sha256, verified_at=_now())
|
||||
|
||||
|
||||
def record(
|
||||
session_factory,
|
||||
*,
|
||||
asset_id: str,
|
||||
stage: str,
|
||||
result: CheckpointResult,
|
||||
add: tuple[str, ...] = (),
|
||||
remove: tuple[str, ...] = (),
|
||||
) -> None:
|
||||
"""Persist the projection for ``(asset_id, stage)`` — one current row per pair.
|
||||
|
||||
The row is what makes divergence durable and reviewable rather than a log line
|
||||
that scrolled away.
|
||||
"""
|
||||
with session_factory() as session:
|
||||
row = session.get(ExifProjection, (asset_id, stage))
|
||||
if row is None:
|
||||
row = ExifProjection(asset_id=asset_id, stage=stage, id=str(uuid.uuid4()))
|
||||
session.add(row)
|
||||
row.projection_version = (row.projection_version or 0) + 1
|
||||
row.desired_json = json.dumps({"add": list(add), "remove": list(remove)})
|
||||
row.divergent_fields = json.dumps(list(result.changed_fields))
|
||||
row.result_file_sha256 = result.sha256
|
||||
row.state = result.state
|
||||
row.error_code = result.reason
|
||||
row.verified_at = result.verified_at
|
||||
row.updated_at = _now()
|
||||
session.commit()
|
||||
|
||||
|
||||
def state_for(session_factory, asset_id: str, stage: str) -> str | None:
|
||||
with session_factory() as session:
|
||||
row = session.get(ExifProjection, (asset_id, stage))
|
||||
return row.state if row else None
|
||||
@@ -16,9 +16,7 @@ from __future__ import annotations
|
||||
import hashlib
|
||||
from pathlib import Path
|
||||
|
||||
from PIL import ImageOps
|
||||
|
||||
from photo_pipeline import imaging
|
||||
from PIL import Image, ImageOps
|
||||
|
||||
PIXEL_HASH_VERSION = 1
|
||||
PHASH_VERSION = 1
|
||||
@@ -43,9 +41,7 @@ def _digest_file(path: Path | str, digest) -> str:
|
||||
|
||||
|
||||
def pixel_sha256(path: Path | str) -> str:
|
||||
# Bounded decode: a header claiming a billion pixels is refused before it is
|
||||
# ever turned into memory (US07-03).
|
||||
with imaging.open_image(path) as image:
|
||||
with Image.open(path) as image:
|
||||
oriented = ImageOps.exif_transpose(image)
|
||||
rgb = oriented.convert("RGB")
|
||||
header = f"{PIXEL_HASH_VERSION}:{rgb.width}x{rgb.height}:".encode()
|
||||
@@ -72,9 +68,7 @@ def phash(path: Path | str) -> str:
|
||||
import numpy as np
|
||||
from scipy.fftpack import dct
|
||||
|
||||
from PIL import Image
|
||||
|
||||
with imaging.open_image(path) as image:
|
||||
with Image.open(path) as image:
|
||||
small = image.convert("L").resize((32, 32), Image.LANCZOS)
|
||||
matrix = np.asarray(small, dtype=np.float64)
|
||||
transformed = dct(dct(matrix, axis=0), axis=1)
|
||||
|
||||
@@ -1,220 +0,0 @@
|
||||
"""Import the last path-keyed CSV state into the database (US07-01).
|
||||
|
||||
``nsfwtag`` cached its safety scores in ``nsfw_scores.csv`` next to the library:
|
||||
one ``path,nsfw_score`` row per photo, four decimals, unreadable rows dropped. That
|
||||
file stops being a source of truth when the CLI is archived, so its scores are
|
||||
imported once — as scored-but-unreviewed ``safety_reviews`` rows on the stable
|
||||
``assets.id`` each path resolves to — and the CSV is left untouched on disk as its
|
||||
own backup.
|
||||
|
||||
The import is deliberately conservative, because a score is evidence about a photo
|
||||
and a path is not an identity:
|
||||
|
||||
- a row whose path matches no known asset is **unmatched**, never a new asset;
|
||||
- an asset a human already reviewed is **never** touched: a score is evidence, a
|
||||
decision is a judgement, and an import may not overwrite the second with the
|
||||
first — the difference is reported as a conflict instead;
|
||||
- a row for an asset that already carries a score is **skipped** unless
|
||||
``overwrite`` is asked for, and a differing score is reported either way;
|
||||
- an unparsable score or a duplicate path is **rejected** with its reason;
|
||||
- rerunning changes nothing that is already imported.
|
||||
|
||||
Everything it did — and everything it refused to do — comes back as a
|
||||
reconciliation report, which the caller can persist next to the database. The
|
||||
donor's own reader is the specification for the format (donor ledger:
|
||||
``nt-score-cache``).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import csv
|
||||
import json
|
||||
import uuid
|
||||
from collections import Counter
|
||||
from dataclasses import asdict, dataclass, field
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
|
||||
from photo_pipeline.models import Asset, AssetPath, SafetyReview
|
||||
|
||||
REPORT_VERSION = 1
|
||||
# The donor wrote scores as four-decimal strings; anything outside 0..1 was never
|
||||
# something it produced, so it is data corruption rather than a score.
|
||||
SCORE_RANGE = (0.0, 1.0)
|
||||
# Recorded as the reviewer so an imported score is never mistaken for a judgement.
|
||||
REVIEWER = "legacy-csv-import"
|
||||
|
||||
|
||||
@dataclass
|
||||
class ImportReport:
|
||||
source: str
|
||||
rows: int = 0
|
||||
imported: int = 0
|
||||
skipped_existing: int = 0
|
||||
unchanged: int = 0
|
||||
reviewed: int = 0
|
||||
unmatched: list[str] = field(default_factory=list)
|
||||
rejected: list[dict] = field(default_factory=list)
|
||||
conflicts: list[dict] = field(default_factory=list)
|
||||
generated_at: str = ""
|
||||
schema_version: int = REPORT_VERSION
|
||||
|
||||
def to_dict(self) -> dict:
|
||||
return asdict(self)
|
||||
|
||||
@property
|
||||
def counts(self) -> dict[str, int]:
|
||||
return {
|
||||
"rows": self.rows,
|
||||
"imported": self.imported,
|
||||
"skipped_existing": self.skipped_existing,
|
||||
"unchanged": self.unchanged,
|
||||
"reviewed": self.reviewed,
|
||||
"unmatched": len(self.unmatched),
|
||||
"rejected": len(self.rejected),
|
||||
"conflicts": len(self.conflicts),
|
||||
}
|
||||
|
||||
|
||||
class LegacyImportService:
|
||||
def __init__(self, session_factory: sessionmaker) -> None:
|
||||
self._session_factory = session_factory
|
||||
|
||||
def import_nsfw_scores(
|
||||
self, csv_path: Path | str, *, overwrite: bool = False, dry_run: bool = False
|
||||
) -> ImportReport:
|
||||
"""Import ``nsfw_scores.csv`` onto asset identity and report what happened."""
|
||||
path = Path(csv_path)
|
||||
report = ImportReport(source=str(path))
|
||||
if not path.is_file():
|
||||
report.rejected.append({"path": str(path), "reason": "csv_missing"})
|
||||
return self._stamp(report)
|
||||
|
||||
with self._session_factory() as session:
|
||||
# Paths are matched against every occurrence an asset ever had, so a
|
||||
# photo scored before a rename is still recognised.
|
||||
by_path = self._path_index(session)
|
||||
latest = self._latest_reviews(session)
|
||||
seen: Counter[str] = Counter()
|
||||
|
||||
for row in self._rows(path, report):
|
||||
report.rows += 1
|
||||
raw_path, raw_score = row
|
||||
seen[raw_path] += 1
|
||||
if seen[raw_path] > 1:
|
||||
report.rejected.append({"path": raw_path, "reason": "duplicate_path"})
|
||||
continue
|
||||
score = _parse_score(raw_score)
|
||||
if score is None:
|
||||
report.rejected.append(
|
||||
{"path": raw_path, "reason": "unparsable_score", "value": raw_score}
|
||||
)
|
||||
continue
|
||||
asset_id = by_path.get(raw_path)
|
||||
if asset_id is None:
|
||||
report.unmatched.append(raw_path)
|
||||
continue
|
||||
|
||||
current = latest.get(asset_id)
|
||||
if current is not None and current.decision:
|
||||
# A human decided this one; the CSV is older evidence.
|
||||
report.reviewed += 1
|
||||
if current.score is None or abs(current.score - score) >= 1e-9:
|
||||
report.conflicts.append(
|
||||
{
|
||||
"path": raw_path,
|
||||
"asset_id": asset_id,
|
||||
"current": current.score,
|
||||
"decision": current.decision,
|
||||
"csv": score,
|
||||
}
|
||||
)
|
||||
continue
|
||||
if current is not None and current.score is not None:
|
||||
if abs(current.score - score) < 1e-9:
|
||||
report.unchanged += 1
|
||||
continue
|
||||
report.conflicts.append(
|
||||
{
|
||||
"path": raw_path,
|
||||
"asset_id": asset_id,
|
||||
"current": current.score,
|
||||
"decision": None,
|
||||
"csv": score,
|
||||
}
|
||||
)
|
||||
if not overwrite:
|
||||
report.skipped_existing += 1
|
||||
continue
|
||||
session.add(
|
||||
SafetyReview(
|
||||
id=str(uuid.uuid4()),
|
||||
asset_id=asset_id,
|
||||
score=score,
|
||||
reviewer=REVIEWER,
|
||||
)
|
||||
)
|
||||
latest[asset_id] = SafetyReview(asset_id=asset_id, score=score)
|
||||
report.imported += 1
|
||||
|
||||
if dry_run:
|
||||
session.rollback()
|
||||
else:
|
||||
session.commit()
|
||||
return self._stamp(report)
|
||||
|
||||
@staticmethod
|
||||
def _rows(path: Path, report: ImportReport):
|
||||
"""Yield ``(path, score)`` pairs, tolerating the donor's own sloppiness."""
|
||||
with open(path, newline="", encoding="utf-8", errors="replace") as handle:
|
||||
for row in csv.DictReader(handle):
|
||||
raw_path = (row.get("path") or "").strip()
|
||||
if not raw_path:
|
||||
report.rejected.append({"path": "", "reason": "missing_path"})
|
||||
continue
|
||||
yield raw_path, row.get("nsfw_score")
|
||||
|
||||
@staticmethod
|
||||
def _latest_reviews(session) -> dict[str, SafetyReview]:
|
||||
"""The current safety row per asset — latest wins, as everywhere else."""
|
||||
latest: dict[str, SafetyReview] = {}
|
||||
for review in session.scalars(select(SafetyReview).order_by(SafetyReview.created_at)):
|
||||
latest[review.asset_id] = review
|
||||
return latest
|
||||
|
||||
@staticmethod
|
||||
def _path_index(session) -> dict[str, str]:
|
||||
"""Every path an asset is or was known by → its stable id."""
|
||||
index: dict[str, str] = {}
|
||||
for asset_id, path in session.execute(select(AssetPath.asset_id, AssetPath.path)):
|
||||
index.setdefault(path, asset_id)
|
||||
for asset_id, path in session.execute(select(Asset.id, Asset.current_path)):
|
||||
if path:
|
||||
index[path] = asset_id # the current path wins over a closed one
|
||||
return index
|
||||
|
||||
@staticmethod
|
||||
def _stamp(report: ImportReport) -> ImportReport:
|
||||
report.generated_at = datetime.now(timezone.utc).isoformat()
|
||||
return report
|
||||
|
||||
|
||||
def write_report(report: ImportReport, directory: Path) -> Path:
|
||||
"""Persist the reconciliation report; the import is not evidence until it is."""
|
||||
directory.mkdir(parents=True, exist_ok=True)
|
||||
path = directory / "legacy-nsfw-import.json"
|
||||
path.write_text(json.dumps(report.to_dict(), indent=2, sort_keys=True), encoding="utf-8")
|
||||
return path
|
||||
|
||||
|
||||
def _parse_score(value) -> float | None:
|
||||
try:
|
||||
score = float(value)
|
||||
except (TypeError, ValueError):
|
||||
return None
|
||||
if not SCORE_RANGE[0] <= score <= SCORE_RANGE[1]:
|
||||
return None
|
||||
return score
|
||||
@@ -108,8 +108,9 @@ from datetime import datetime, timezone
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
|
||||
from photo_pipeline.models import Asset, ExifProjection, SafetyReview
|
||||
from photo_pipeline.services import exif_checkpoint
|
||||
from photo_pipeline.integrations import exiftool
|
||||
from photo_pipeline.models import Asset, SafetyReview
|
||||
from photo_pipeline.services import hashing
|
||||
|
||||
DECISIONS = {SFW, NSFW, "deferred"}
|
||||
|
||||
@@ -173,14 +174,6 @@ class SafetyService:
|
||||
with self._session_factory() as session:
|
||||
assets = list(session.scalars(_eligible_assets_query().order_by(Asset.current_path)))
|
||||
latest = self._latest_by_asset(session)
|
||||
# One query, not one per asset: the reviewer needs to see a divergent
|
||||
# checkpoint, which is neither "verified" nor a plain failure (US07-03).
|
||||
projections = {
|
||||
row.asset_id: row.state
|
||||
for row in session.scalars(
|
||||
select(ExifProjection).where(ExifProjection.stage == "safety")
|
||||
)
|
||||
}
|
||||
rows = []
|
||||
for asset in assets:
|
||||
review = latest.get(asset.id)
|
||||
@@ -196,7 +189,6 @@ class SafetyService:
|
||||
"decision": decision,
|
||||
"suggested": classify(review.score) if review and review.score is not None else None,
|
||||
"exif_verified": bool(review and review.exif_verified_at),
|
||||
"exif_state": projections.get(asset.id),
|
||||
}
|
||||
)
|
||||
return {"total": len(rows), "items": rows[offset : offset + limit]}
|
||||
@@ -255,24 +247,13 @@ class SafetyService:
|
||||
result_sha256 = None
|
||||
if write_exif and decision in (SFW, NSFW) and path:
|
||||
ops = exif_projection(decision)
|
||||
# The full checkpoint: write the owned keyword, read the whole file back,
|
||||
# and prove every field this stage does not own survived. A divergent
|
||||
# result is recorded and left alone — it must not count as verified, so
|
||||
# upload stays blocked until a human decides (US07-03).
|
||||
result = exif_checkpoint.run(
|
||||
path, add=tuple(ops["add"]), remove=tuple(ops["remove"])
|
||||
)
|
||||
exif_checkpoint.record(
|
||||
self._session_factory,
|
||||
asset_id=asset_id,
|
||||
stage="safety",
|
||||
result=result,
|
||||
add=tuple(ops["add"]),
|
||||
remove=tuple(ops["remove"]),
|
||||
)
|
||||
if result.verified:
|
||||
exif_verified_at = result.verified_at
|
||||
result_sha256 = result.sha256
|
||||
if exiftool.apply_keywords(path, add=ops["add"], remove=ops["remove"]):
|
||||
# Read back: the chosen keyword present, the opposite absent.
|
||||
keywords = exiftool.read_keyword_sets([path]).get(path, set())
|
||||
opposite = NSFW if decision == SFW else SFW
|
||||
if decision in keywords and opposite not in keywords:
|
||||
exif_verified_at = _now()
|
||||
result_sha256 = hashing.sha256_file(path)
|
||||
|
||||
now = _now()
|
||||
with self._session_factory() as session:
|
||||
|
||||
@@ -20,16 +20,15 @@ pa-imaging).
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import uuid
|
||||
from pathlib import Path
|
||||
|
||||
from PIL import Image, ImageOps
|
||||
from PIL import Image, ImageOps, UnidentifiedImageError
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.exc import IntegrityError
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
|
||||
from photo_pipeline import imaging, path_policy
|
||||
from photo_pipeline import path_policy
|
||||
from photo_pipeline.config import Config
|
||||
from photo_pipeline.models import Asset, Thumbnail
|
||||
from photo_pipeline.services import availability
|
||||
@@ -43,9 +42,6 @@ except Exception: # pragma: no cover
|
||||
pass
|
||||
|
||||
SIZES = (256, 512, 1280)
|
||||
# Every in-flight render writes one of these beside its destination; startup
|
||||
# cleanup recognises exactly this pattern and nothing else.
|
||||
TEMP_SUFFIX = ".tmp"
|
||||
THUMB_VERSION = 1
|
||||
THUMB_FORMAT = "webp"
|
||||
# The size kept as durable comparison evidence for archived assets (concept §9).
|
||||
@@ -133,10 +129,7 @@ class ThumbnailService:
|
||||
raise ThumbnailUnavailable(f"asset {asset_id} has no readable file")
|
||||
source = str(source)
|
||||
if source == asset.current_path:
|
||||
# Render the *resolved* path the check approved: revalidating and then
|
||||
# reopening the original name would let a symlink swapped in between
|
||||
# the two steps decide which bytes are served (US07-02).
|
||||
source = str(self._validate_path(source)) # archive roots lie outside
|
||||
self._validate_path(source) # archive roots lie outside the library
|
||||
|
||||
# Rendering happens outside the DB session (no transaction held during I/O).
|
||||
try:
|
||||
@@ -202,16 +195,20 @@ class ThumbnailService:
|
||||
session.commit()
|
||||
|
||||
# ── path safety ──────────────────────────────────────────────────────────
|
||||
def _validate_path(self, current_path: str) -> Path:
|
||||
"""The resolved path to read, or ``PathNotAllowed``.
|
||||
|
||||
The message names no path: a refusal is returned to the browser, and where
|
||||
the library lives is not the caller's business (US07-02).
|
||||
"""
|
||||
try:
|
||||
return path_policy.resolve_in_roots(self._config.library_roots, current_path)
|
||||
except path_policy.PathPolicyError as error:
|
||||
raise PathNotAllowed(str(error)) from None
|
||||
def _validate_path(self, current_path: str) -> None:
|
||||
path = Path(current_path)
|
||||
if path_policy.is_excluded(path):
|
||||
raise PathNotAllowed(f"excluded path: {current_path}")
|
||||
roots = self._config.library_roots
|
||||
if not roots:
|
||||
return
|
||||
for root in roots:
|
||||
try:
|
||||
path_policy.resolve_within(Path(root), path)
|
||||
return
|
||||
except path_policy.PathPolicyError:
|
||||
continue
|
||||
raise PathNotAllowed(f"path outside configured roots: {current_path}")
|
||||
|
||||
# ── cache key + rendering ──────────────────────────────────────────────────
|
||||
@staticmethod
|
||||
@@ -226,33 +223,28 @@ class ThumbnailService:
|
||||
return self._cache_dir / safe[:2] / f"{safe}.{THUMB_FORMAT}"
|
||||
|
||||
def _render(self, source: str, size: int, cache_key: str) -> dict:
|
||||
"""Decode bounded, orient, colour-manage, resize, and write atomically.
|
||||
|
||||
The temporary file is removed on every failure path: a decoder that dies
|
||||
halfway through ``save`` would otherwise leave a stray ``.tmp`` in the cache
|
||||
forever (US07-03).
|
||||
"""
|
||||
destination = self._cache_path(cache_key)
|
||||
destination.parent.mkdir(parents=True, exist_ok=True)
|
||||
tmp = destination.with_name(f".{destination.name}.{uuid.uuid4().hex}{TEMP_SUFFIX}")
|
||||
try:
|
||||
with imaging.open_image(source, max_pixels=self._config.thumbnail_max_pixels) as image:
|
||||
imaging.draft(image, size) # JPEG decodes near the target size
|
||||
with Image.open(source) as image:
|
||||
width, height = image.size
|
||||
if width * height > self._config.thumbnail_max_pixels:
|
||||
raise ImageTooLarge(
|
||||
f"{width}x{height} exceeds {self._config.thumbnail_max_pixels} px"
|
||||
)
|
||||
oriented = ImageOps.exif_transpose(image)
|
||||
mode = "RGBA" if _has_alpha(oriented) else "RGB"
|
||||
converted = imaging.to_srgb(oriented, mode=mode)
|
||||
converted = oriented.convert(mode)
|
||||
converted.thumbnail((size, size), Image.LANCZOS)
|
||||
out_width, out_height = converted.size
|
||||
destination = self._cache_path(cache_key)
|
||||
destination.parent.mkdir(parents=True, exist_ok=True)
|
||||
tmp = destination.with_name(f".{destination.name}.{uuid.uuid4().hex}.tmp")
|
||||
converted.save(tmp, format="WEBP", quality=82, method=4)
|
||||
except imaging.ImageTooLarge as error:
|
||||
tmp.unlink(missing_ok=True)
|
||||
raise ImageTooLarge(str(error)) from None
|
||||
except imaging.UndecodableImage as error:
|
||||
tmp.unlink(missing_ok=True)
|
||||
raise UnsupportedImage(str(error)) from None
|
||||
except Exception:
|
||||
tmp.unlink(missing_ok=True)
|
||||
except ImageTooLarge:
|
||||
raise
|
||||
except (UnidentifiedImageError, OSError, ValueError) as error:
|
||||
raise UnsupportedImage(f"cannot decode {source}: {error}") from error
|
||||
|
||||
import os
|
||||
|
||||
os.replace(tmp, destination)
|
||||
return {
|
||||
@@ -262,28 +254,6 @@ class ThumbnailService:
|
||||
"format": THUMB_FORMAT,
|
||||
}
|
||||
|
||||
def cleanup_temp_files(self) -> int:
|
||||
"""Remove leftover render temporaries, and only those.
|
||||
|
||||
Runs at startup, where the concept is explicit: clean *recognized* stale
|
||||
temporary files, never arbitrary ones. The pattern is this service's own —
|
||||
a dot-prefixed name inside the managed cache directory ending in
|
||||
``.tmp`` — so nothing outside the cache and nothing a user put there can
|
||||
match. Returns how many were removed.
|
||||
"""
|
||||
if not self._cache_dir.is_dir():
|
||||
return 0
|
||||
removed = 0
|
||||
for leftover in self._cache_dir.rglob(f".*{TEMP_SUFFIX}"):
|
||||
if not leftover.is_file() or leftover.is_symlink():
|
||||
continue
|
||||
try:
|
||||
leftover.unlink()
|
||||
removed += 1
|
||||
except OSError: # pragma: no cover - a racing render already won
|
||||
pass
|
||||
return removed
|
||||
|
||||
# ── persistence ────────────────────────────────────────────────────────────
|
||||
def _record_ready(
|
||||
self, cache_key: str, asset_id: str, size: int, rendered: dict, *, protected: bool = False
|
||||
|
||||
@@ -24,8 +24,6 @@ test = [
|
||||
|
||||
[tool.ruff]
|
||||
line-length = 100
|
||||
# The frozen CLI archive is evidence, not code under maintenance (US07-01).
|
||||
extend-exclude = ["legacy_cli_archive"]
|
||||
|
||||
[tool.pytest.ini_options]
|
||||
testpaths = ["tests"]
|
||||
|
||||
@@ -14,11 +14,6 @@ import pytest
|
||||
from PIL import Image
|
||||
|
||||
REPO = Path(__file__).resolve().parents[2]
|
||||
# The donors are frozen in the read-only archive (US07-01). Only this suite — and
|
||||
# the parity test that compares against them — puts that directory on sys.path;
|
||||
# production never does, which tests/unit/test_legacy_archive.py enforces.
|
||||
ARCHIVED_SOURCES = REPO / "legacy_cli_archive" / "src"
|
||||
sys.path.insert(0, str(ARCHIVED_SOURCES))
|
||||
sys.path.insert(0, str(REPO))
|
||||
|
||||
EXIFTOOL = shutil.which("exiftool")
|
||||
|
||||
@@ -1,20 +1,13 @@
|
||||
"""Ledger lint (US01-01, extended by US07-01).
|
||||
|
||||
Every donor-ledger row must carry a real source reference — now inside the frozen
|
||||
archive — a target location, and either existing characterization test IDs or a
|
||||
real pending backlog story. Since archival (US07-01) a row may also be ``resolved``:
|
||||
its replacement has shipped, ``parity`` names tests that exist and prove it, and
|
||||
``delta`` states every intentional difference. Nothing may quietly become
|
||||
"finished" without one of those two."""
|
||||
"""Ledger lint (US01-01): every donor-ledger row must carry a real source
|
||||
reference, a target location, and either existing characterization test IDs or
|
||||
a real pending backlog story."""
|
||||
import re
|
||||
from pathlib import Path
|
||||
|
||||
import yaml
|
||||
|
||||
REPO = Path(__file__).resolve().parents[2]
|
||||
ARCHIVE = REPO / "legacy_cli_archive"
|
||||
LEDGER = ARCHIVE / "donor_ledger.yaml"
|
||||
ARCHIVED_SOURCES = ARCHIVE / "src"
|
||||
LEDGER = REPO / "donor_ledger.yaml"
|
||||
STORIES = REPO / "delivery_backlog" / "stories"
|
||||
TESTS_DIR = Path(__file__).resolve().parent
|
||||
|
||||
@@ -22,7 +15,7 @@ CLASSIFICATIONS = {"reuse", "extract", "refactor", "replace"}
|
||||
REQUIRED_AREAS = {"discovery", "hashing", "imaging", "nsfw", "vision", "exif",
|
||||
"database", "ui", "configuration", "logging", "cancellation",
|
||||
"error"}
|
||||
STATUSES = {"characterized", "resolved", "pending"}
|
||||
STATUSES = {"characterized", "pending"}
|
||||
|
||||
|
||||
def load_rows():
|
||||
@@ -56,11 +49,10 @@ def test_rows_have_required_fields_and_unique_ids():
|
||||
|
||||
|
||||
def test_source_references_resolve():
|
||||
"""Source paths are relative to the archive: the donors moved there, whole."""
|
||||
for r in load_rows():
|
||||
src = r["source"]
|
||||
f = ARCHIVED_SOURCES / src["file"]
|
||||
assert f.is_file(), f"{r['id']}: source file {src['file']} missing from the archive"
|
||||
f = REPO / src["file"]
|
||||
assert f.is_file(), f"{r['id']}: source file {src['file']} missing"
|
||||
text = f.read_text(encoding="utf-8")
|
||||
for sym in src["symbols"]:
|
||||
assert sym in text, f"{r['id']}: symbol {sym!r} not found in {src['file']}"
|
||||
@@ -71,9 +63,7 @@ def test_rows_have_tests_or_pending_story():
|
||||
for r in load_rows():
|
||||
tests = r.get("tests", [])
|
||||
pending = r.get("pending_story")
|
||||
parity = r.get("parity", [])
|
||||
assert tests or pending or parity, \
|
||||
f"{r['id']}: neither characterization tests, parity tests, nor a pending story"
|
||||
assert tests or pending, f"{r['id']}: neither tests nor pending_story"
|
||||
for t in tests:
|
||||
assert t in known_tests, f"{r['id']}: unknown test id {t}"
|
||||
if pending:
|
||||
@@ -83,58 +73,16 @@ def test_rows_have_tests_or_pending_story():
|
||||
assert tests, f"{r['id']}: characterized rows need test ids"
|
||||
|
||||
|
||||
def test_resolved_rows_name_their_parity_or_their_delta():
|
||||
"""A resolved row is a claim that the behavior is handled. It has to say how:
|
||||
tests that prove the replacement, or a stated difference — usually both."""
|
||||
for r in load_rows():
|
||||
if r["status"] != "resolved":
|
||||
assert "parity" not in r, f"{r['id']}: parity on a non-resolved row"
|
||||
continue
|
||||
parity = r.get("parity", [])
|
||||
delta = r.get("delta")
|
||||
assert parity or delta, f"{r['id']}: resolved without parity tests or a delta"
|
||||
if delta:
|
||||
assert len(str(delta).strip()) >= 20, f"{r['id']}: delta too thin to be a reason"
|
||||
for ref in parity:
|
||||
rel, _, func = ref.partition("::")
|
||||
path = REPO / rel
|
||||
assert path.is_file(), f"{r['id']}: parity test file {rel} missing"
|
||||
assert f"def {func}" in path.read_text(encoding="utf-8"), \
|
||||
f"{r['id']}: parity test {ref} not found"
|
||||
|
||||
|
||||
def test_pending_rows_are_the_only_unfinished_work():
|
||||
"""The ledger is the honest list of what has not been carried over: a pending
|
||||
row names the story that will, and that story must still be open work."""
|
||||
for r in load_rows():
|
||||
if r["status"] != "pending":
|
||||
continue
|
||||
story = r.get("pending_story")
|
||||
assert story, f"{r['id']}: pending without a story"
|
||||
assert list(STORIES.glob(f"{story}-*.md")), f"{r['id']}: unknown story {story}"
|
||||
|
||||
|
||||
def test_every_target_module_exists():
|
||||
"""A row is only finished if the thing it points at is really there."""
|
||||
for r in load_rows():
|
||||
if r["status"] == "pending":
|
||||
continue
|
||||
modules = re.findall(r"photo_pipeline/[\w/]+\.py", str(r["target"]))
|
||||
for module in modules:
|
||||
assert (REPO / module).is_file(), f"{r['id']}: target {module} does not exist"
|
||||
|
||||
|
||||
def test_all_required_areas_covered():
|
||||
covered = {r["area"] for r in load_rows()}
|
||||
assert REQUIRED_AREAS <= covered, f"uncovered areas: {REQUIRED_AREAS - covered}"
|
||||
assert covered <= REQUIRED_AREAS, f"unknown areas: {covered - REQUIRED_AREAS}"
|
||||
|
||||
|
||||
def test_every_donor_is_in_the_archive_and_nowhere_else():
|
||||
"""US01-01 forbade archiving before characterization; US07-01 requires it after.
|
||||
Each donor exists exactly once — frozen, in the archive."""
|
||||
def test_no_legacy_file_moved():
|
||||
# US01-01 explicitly forbids moving/archiving donors; the ledger's source
|
||||
# files must all still exist at their original locations.
|
||||
for donor in ("photo_analyzer.py", "nsfwtag/scoring.py", "nsfwtag/exif.py",
|
||||
"nsfwtag/server.py", "webapp/query.py", "webapp/runner.py",
|
||||
"webapp/server.py"):
|
||||
assert (ARCHIVED_SOURCES / donor).is_file(), f"donor missing from archive: {donor}"
|
||||
assert not (REPO / donor).exists(), f"donor still live at the repo root: {donor}"
|
||||
assert (REPO / donor).is_file(), f"donor moved: {donor}"
|
||||
|
||||
@@ -1,15 +1,4 @@
|
||||
"""Make the repository root importable for the pipeline test suites, and give every
|
||||
suite the application session the API requires since US07-02.
|
||||
|
||||
The suites drive the API the way the browser does — module-level ``httpx`` calls and
|
||||
``TestClient`` — so instead of threading a cookie through several hundred call sites,
|
||||
both clients bootstrap the session themselves exactly like ``frontend/js/api.js``:
|
||||
fetch ``/api/v1/session`` once, then send the cookie plus the CSRF header, and
|
||||
re-bootstrap once on 401 (a restarted server issues a new session).
|
||||
|
||||
Security tests deliberately bypass this by constructing their own ``httpx.Client``;
|
||||
only the module-level helpers are wrapped.
|
||||
"""
|
||||
"""Make the repository root importable for the pipeline test suites."""
|
||||
|
||||
import sys
|
||||
from pathlib import Path
|
||||
@@ -17,85 +6,3 @@ from pathlib import Path
|
||||
REPO = Path(__file__).resolve().parents[1]
|
||||
if str(REPO) not in sys.path:
|
||||
sys.path.insert(0, str(REPO))
|
||||
|
||||
import httpx
|
||||
import httpx._api # httpx.get/post resolve `request` in this module
|
||||
import pytest
|
||||
from starlette.testclient import TestClient
|
||||
|
||||
from photo_pipeline.api.security import CSRF_HEADER, SESSION_COOKIE
|
||||
|
||||
_SESSIONS: dict[str, tuple[str, str]] = {} # base url -> (session id, csrf token)
|
||||
|
||||
|
||||
def _base(url) -> str:
|
||||
parts = httpx.URL(str(url))
|
||||
return f"{parts.scheme}://{parts.netloc.decode()}"
|
||||
|
||||
|
||||
def _bootstrap(base: str) -> tuple[str, str]:
|
||||
if base not in _SESSIONS:
|
||||
response = httpx.Client(timeout=10).get(f"{base}/api/v1/session")
|
||||
_SESSIONS[base] = (
|
||||
response.cookies.get(SESSION_COOKIE, ""),
|
||||
response.json().get("csrf_token", ""),
|
||||
)
|
||||
return _SESSIONS[base]
|
||||
|
||||
|
||||
def session_client(base_url: str, **kwargs) -> httpx.Client:
|
||||
"""An ``httpx.Client`` that has bootstrapped its own application session.
|
||||
|
||||
For suites that keep one client for a whole journey; it holds its own cookie, so
|
||||
a client made after a server restart picks up the new session by construction.
|
||||
"""
|
||||
client = httpx.Client(base_url=base_url, **kwargs)
|
||||
client.headers[CSRF_HEADER] = client.get("/api/v1/session").json()["csrf_token"]
|
||||
return client
|
||||
|
||||
|
||||
def _authorized(url, headers, cookies):
|
||||
session_id, token = _bootstrap(_base(url))
|
||||
headers = dict(headers or {})
|
||||
headers.setdefault(CSRF_HEADER, token)
|
||||
cookies = dict(cookies or {})
|
||||
cookies.setdefault(SESSION_COOKIE, session_id)
|
||||
return headers, cookies
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True, scope="session")
|
||||
def _api_session():
|
||||
real_request, real_stream = httpx._api.request, httpx._api.stream
|
||||
real_client_request, real_client_init = TestClient.request, TestClient.__init__
|
||||
|
||||
def request(method, url, *, headers=None, cookies=None, **kwargs):
|
||||
sent, jar = _authorized(url, headers, cookies)
|
||||
response = real_request(method, url, headers=sent, cookies=jar, **kwargs)
|
||||
if response.status_code == 401:
|
||||
_SESSIONS.pop(_base(url), None)
|
||||
sent, jar = _authorized(url, headers, cookies)
|
||||
response = real_request(method, url, headers=sent, cookies=jar, **kwargs)
|
||||
return response
|
||||
|
||||
def stream(method, url, *, headers=None, cookies=None, **kwargs):
|
||||
sent, jar = _authorized(url, headers, cookies)
|
||||
return real_stream(method, url, headers=sent, cookies=jar, **kwargs)
|
||||
|
||||
def client_init(self, app, *args, base_url="http://127.0.0.1", **kwargs):
|
||||
# The default "http://testserver" is not a local host, which is exactly what
|
||||
# the Host check refuses; in-process tests are still same-origin callers.
|
||||
real_client_init(self, app, *args, base_url=base_url, **kwargs)
|
||||
|
||||
def client_request(self, method, url, *, headers=None, **kwargs):
|
||||
if CSRF_HEADER not in self.headers:
|
||||
response = real_client_request(self, "GET", "/api/v1/session")
|
||||
self.headers[CSRF_HEADER] = response.json()["csrf_token"]
|
||||
return real_client_request(self, method, url, headers=headers, **kwargs)
|
||||
|
||||
httpx._api.request, httpx.request = request, request
|
||||
httpx._api.stream, httpx.stream = stream, stream
|
||||
TestClient.request, TestClient.__init__ = client_request, client_init
|
||||
yield
|
||||
httpx._api.request, httpx.request = real_request, real_request
|
||||
httpx._api.stream, httpx.stream = real_stream, real_stream
|
||||
TestClient.request, TestClient.__init__ = real_client_request, real_client_init
|
||||
|
||||
@@ -22,8 +22,6 @@ import numpy as np
|
||||
import pytest
|
||||
from PIL import Image
|
||||
|
||||
from tests.conftest import session_client
|
||||
|
||||
REPO = Path(__file__).resolve().parents[2]
|
||||
|
||||
|
||||
@@ -94,7 +92,7 @@ class ServerController:
|
||||
pytest.fail(f"server exited: {err.decode(errors='replace')}")
|
||||
try:
|
||||
if httpx.get(f"{self.base}/api/v1/health/ready", timeout=1).status_code == 200:
|
||||
self.client = session_client(self.base, timeout=10)
|
||||
self.client = httpx.Client(base_url=self.base, timeout=10)
|
||||
return
|
||||
except httpx.HTTPError:
|
||||
time.sleep(0.2)
|
||||
|
||||
@@ -17,9 +17,6 @@ import httpx
|
||||
import numpy as np
|
||||
import pytest
|
||||
from PIL import Image
|
||||
from playwright.sync_api import expect
|
||||
|
||||
from tests.conftest import session_client
|
||||
|
||||
REPO = Path(__file__).resolve().parents[2]
|
||||
|
||||
@@ -114,7 +111,7 @@ def server(tmp_path):
|
||||
proc.terminate()
|
||||
pytest.fail("server never became ready")
|
||||
|
||||
client = session_client(base, timeout=5)
|
||||
client = httpx.Client(base_url=base, timeout=5)
|
||||
|
||||
def cluster_by_method(method):
|
||||
clusters = client.get("/api/v1/duplicates/clusters").json()["items"]
|
||||
@@ -151,9 +148,9 @@ def test_fuzzy_decision_requires_confirmation(page, server):
|
||||
page.get_by_test_id("confirm").wait_for()
|
||||
assert "open" in page.get_by_test_id("cluster-state").inner_text()
|
||||
page.get_by_test_id("confirm-yes").click()
|
||||
# A locator assertion, not wait_for_function: the app's CSP forbids eval, and
|
||||
# Playwright's polling predicate is evaluated as a string in the page (US07-02).
|
||||
expect(page.get_by_test_id("cluster-state")).to_contain_text("dismissed")
|
||||
page.wait_for_function(
|
||||
"document.querySelector('[data-testid=cluster-state]').innerText.includes('dismissed')"
|
||||
)
|
||||
|
||||
|
||||
def test_stale_version_shows_conflict(page, server):
|
||||
@@ -179,9 +176,9 @@ def test_decision_persists_after_reload(page, server):
|
||||
page.goto(f"{server.base}/app/#/duplicates/{cluster['id']}")
|
||||
page.get_by_test_id("not-duplicate").click()
|
||||
page.get_by_test_id("confirm-yes").click()
|
||||
# A locator assertion, not wait_for_function: the app's CSP forbids eval, and
|
||||
# Playwright's polling predicate is evaluated as a string in the page (US07-02).
|
||||
expect(page.get_by_test_id("cluster-state")).to_contain_text("dismissed")
|
||||
page.wait_for_function(
|
||||
"document.querySelector('[data-testid=cluster-state]').innerText.includes('dismissed')"
|
||||
)
|
||||
page.reload()
|
||||
page.get_by_test_id("cluster-state").wait_for()
|
||||
assert "dismissed" in page.get_by_test_id("cluster-state").inner_text()
|
||||
|
||||
@@ -1,280 +0,0 @@
|
||||
"""US07-02 black-box security tests: authorization and path boundaries.
|
||||
|
||||
Everything here talks to a real ``photo_pipeline serve`` child process over HTTP with
|
||||
its own ``httpx.Client``, deliberately outside the session helper the rest of the
|
||||
suite uses — an attacker does not get a bootstrapped client.
|
||||
|
||||
The threat is a page in the user's browser, not a remote attacker: the app listens on
|
||||
127.0.0.1, so any site the user visits can send requests to it and can point an
|
||||
``<img>`` at its media endpoints. Each journey below is one of those attempts.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from contextlib import contextmanager
|
||||
from pathlib import Path
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
|
||||
from tests.e2e._pipeline_harness import Server, image, seed_library
|
||||
|
||||
TIMEOUT = 20
|
||||
SENTINEL_KEY = "immich-sentinel-9f3a2b"
|
||||
SESSION_COOKIE = "pp_session"
|
||||
CSRF_HEADER = "X-CSRF-Token"
|
||||
|
||||
# A GET, a mutation, and a media endpoint: the three shapes the policy must cover.
|
||||
PROTECTED = [
|
||||
("GET", "/api/v1/workflow", None),
|
||||
("POST", "/api/v1/albums/proposals", {}),
|
||||
("GET", "/api/v1/assets/{asset}/thumbnail?size=256", None),
|
||||
]
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def stack(tmp_path_factory):
|
||||
"""One server, one album of two photos, and a credential sentinel in its config."""
|
||||
tmp_path = tmp_path_factory.mktemp("security")
|
||||
seeded = seed_library(tmp_path, {"a": 1, "b": 2}, {"a": "sfw", "b": "sfw"})
|
||||
outside = tmp_path / "outside"
|
||||
outside.mkdir()
|
||||
image(outside / "secret.jpg", 99)
|
||||
server = Server(
|
||||
seeded,
|
||||
extra_env={
|
||||
"PHOTO_PIPELINE_IMMICH_API_KEY": SENTINEL_KEY,
|
||||
"PHOTO_PIPELINE_IMMICH_SERVER_URL": "http://127.0.0.1:1",
|
||||
},
|
||||
).start()
|
||||
yield server, seeded, outside
|
||||
server.stop()
|
||||
|
||||
|
||||
@contextmanager
|
||||
def anonymous(server):
|
||||
with httpx.Client(base_url=server.base, timeout=TIMEOUT) as client:
|
||||
yield client
|
||||
|
||||
|
||||
@contextmanager
|
||||
def authenticated(server):
|
||||
"""A browser that has loaded the app: session cookie in the jar, token in a header."""
|
||||
with httpx.Client(base_url=server.base, timeout=TIMEOUT) as client:
|
||||
client.headers[CSRF_HEADER] = client.get("/api/v1/session").json()["csrf_token"]
|
||||
yield client
|
||||
|
||||
|
||||
def call(client, method, path, body, asset):
|
||||
return client.request(method, path.format(asset=asset), json=body)
|
||||
|
||||
|
||||
def test_the_api_refuses_every_caller_without_a_session(stack):
|
||||
server, seeded, _ = stack
|
||||
with anonymous(server) as client:
|
||||
for method, path, body in PROTECTED:
|
||||
response = call(client, method, path, body, seeded.asset_ids["a"])
|
||||
assert response.status_code == 401, path
|
||||
assert response.json()["error"]["code"] == "unauthenticated"
|
||||
# Liveness and readiness stay open: an orchestrator holds no session.
|
||||
assert client.get("/api/v1/health/ready").status_code == 200
|
||||
assert client.get("/api/v1/health/live").status_code == 200
|
||||
|
||||
|
||||
def test_a_guessed_session_cookie_is_refused(stack):
|
||||
server, _, _ = stack
|
||||
with anonymous(server) as client:
|
||||
client.cookies.set(SESSION_COOKIE, "guessed", domain="127.0.0.1")
|
||||
response = client.get("/api/v1/workflow")
|
||||
assert response.status_code == 401
|
||||
assert response.json()["error"]["code"] == "unauthenticated"
|
||||
|
||||
|
||||
def test_the_bootstrap_issues_a_strict_httponly_cookie(stack):
|
||||
server, _, _ = stack
|
||||
with anonymous(server) as client:
|
||||
response = client.get("/api/v1/session")
|
||||
cookie = response.headers["set-cookie"].lower()
|
||||
assert "httponly" in cookie and "samesite=strict" in cookie and "path=/" in cookie
|
||||
assert response.json()["csrf_token"]
|
||||
# The token is in the body, which no other origin may read: no CORS header
|
||||
# grants access to it.
|
||||
assert "access-control-allow-origin" not in response.headers
|
||||
|
||||
|
||||
def test_a_session_without_the_csrf_token_may_read_but_not_mutate(stack):
|
||||
server, _, _ = stack
|
||||
with authenticated(server) as client:
|
||||
del client.headers[CSRF_HEADER]
|
||||
assert client.get("/api/v1/albums/proposals").status_code == 200
|
||||
|
||||
response = client.post("/api/v1/albums/proposals", json={})
|
||||
assert response.status_code == 403
|
||||
assert response.json()["error"]["code"] == "csrf_failed"
|
||||
|
||||
response = client.post(
|
||||
"/api/v1/albums/proposals", json={}, headers={CSRF_HEADER: "guessed"}
|
||||
)
|
||||
assert response.status_code == 403
|
||||
# And nothing was created behind the refusal.
|
||||
with authenticated(server) as client:
|
||||
assert client.get("/api/v1/albums/proposals").json()["items"] == []
|
||||
|
||||
|
||||
def test_a_foreign_origin_cannot_mutate_even_with_a_session(stack):
|
||||
server, _, _ = stack
|
||||
with authenticated(server) as client:
|
||||
for origin in ("http://evil.example", "http://127.0.0.1:1", "null"):
|
||||
response = client.post(
|
||||
"/api/v1/albums/proposals", json={}, headers={"Origin": origin}
|
||||
)
|
||||
assert response.status_code == 403, origin
|
||||
assert response.json()["error"]["code"] == "origin_not_allowed"
|
||||
# This app's own origin is accepted, so the check is not simply refusing all.
|
||||
allowed = client.post(
|
||||
"/api/v1/albums/proposals", json={}, headers={"Origin": server.base}
|
||||
)
|
||||
assert allowed.status_code == 200
|
||||
|
||||
|
||||
def test_a_rebinding_host_is_refused(stack):
|
||||
"""A name that resolves to 127.0.0.1 makes the browser treat the attacker's page
|
||||
as same-origin. The Host header still carries that name, so it is checked."""
|
||||
server, _, _ = stack
|
||||
with authenticated(server) as client:
|
||||
response = client.get("/api/v1/workflow", headers={"Host": "photos.evil.example"})
|
||||
assert response.status_code == 403
|
||||
assert response.json()["error"]["code"] == "host_not_allowed"
|
||||
|
||||
|
||||
def test_media_cannot_be_embedded_by_another_page(stack):
|
||||
server, seeded, _ = stack
|
||||
url = f"/api/v1/assets/{seeded.asset_ids['a']}/thumbnail?size=256"
|
||||
with authenticated(server) as client:
|
||||
# What an <img> on another site produces: no Origin, but a cross-site marker.
|
||||
blocked = client.get(url, headers={"Sec-Fetch-Site": "cross-site"})
|
||||
assert blocked.status_code == 403
|
||||
assert blocked.json()["error"]["code"] == "cross_site_blocked"
|
||||
|
||||
served = client.get(url, headers={"Sec-Fetch-Site": "same-origin"})
|
||||
assert served.status_code == 200
|
||||
assert served.headers["content-type"] == "image/webp"
|
||||
assert served.headers["cross-origin-resource-policy"] == "same-origin"
|
||||
# Photos must never land in a shared cache.
|
||||
assert served.headers["cache-control"].startswith("private")
|
||||
|
||||
|
||||
def test_every_response_carries_the_default_headers_and_no_cors(stack):
|
||||
server, seeded, _ = stack
|
||||
with authenticated(server) as client:
|
||||
responses = [
|
||||
client.get("/api/v1/workflow"),
|
||||
client.get(f"/api/v1/assets/{seeded.asset_ids['a']}/thumbnail?size=256"),
|
||||
client.get("/app/"),
|
||||
client.get("/api/v1/does-not-exist"),
|
||||
]
|
||||
for response in responses:
|
||||
assert response.headers["x-content-type-options"] == "nosniff"
|
||||
assert response.headers["x-frame-options"] == "DENY"
|
||||
assert response.headers["referrer-policy"] == "no-referrer"
|
||||
assert "frame-ancestors 'none'" in response.headers["content-security-policy"]
|
||||
assert "access-control-allow-origin" not in response.headers
|
||||
assert "access-control-allow-credentials" not in response.headers
|
||||
|
||||
|
||||
def test_a_traversal_attempt_addresses_nothing(stack):
|
||||
"""Identifiers are database keys, not paths: traversal has nowhere to land."""
|
||||
server, _, _ = stack
|
||||
with authenticated(server) as client:
|
||||
for path in (
|
||||
"/api/v1/albums/proposals/..%2F..%2F..%2Fetc%2Fpasswd",
|
||||
"/api/v1/albums/proposals/../../../etc/passwd",
|
||||
"/api/v1/assets/..%2F..%2Fetc%2Fpasswd/thumbnail?size=256",
|
||||
"/api/v1/assets/%2Fetc%2Fpasswd/thumbnail?size=256",
|
||||
):
|
||||
response = client.get(path)
|
||||
assert response.status_code in (404, 422), path
|
||||
assert "root:" not in response.text, path
|
||||
|
||||
|
||||
def test_an_oversized_request_is_refused_before_it_is_parsed(stack):
|
||||
server, _, _ = stack
|
||||
with authenticated(server) as client:
|
||||
response = client.post(
|
||||
"/api/v1/albums/proposals",
|
||||
content=json.dumps({"albums": ["x" * 2_000_000]}),
|
||||
headers={"Content-Type": "application/json"},
|
||||
)
|
||||
assert response.status_code == 413
|
||||
assert response.json()["error"]["code"] == "payload_too_large"
|
||||
|
||||
|
||||
def test_a_malformed_request_reports_the_field_and_nothing_else(stack):
|
||||
server, _, _ = stack
|
||||
with authenticated(server) as client:
|
||||
for content in ("{", '{"albums": 5}', ""):
|
||||
response = client.post(
|
||||
"/api/v1/albums/proposals",
|
||||
content=content,
|
||||
headers={"Content-Type": "application/json"},
|
||||
)
|
||||
assert response.status_code == 422, content
|
||||
error = response.json()["error"]
|
||||
assert error["code"] == "invalid_request"
|
||||
assert "Traceback" not in response.text and "photo_pipeline/" not in response.text
|
||||
|
||||
|
||||
def test_errors_reveal_neither_the_credential_nor_an_internal(stack):
|
||||
"""The API key is configured but must appear nowhere; failures additionally say
|
||||
nothing about where the library lives or how the server is built.
|
||||
|
||||
Successful responses are a different matter: path previews are the point of the
|
||||
rename and archive views, and the operator is the one who owns those paths.
|
||||
"""
|
||||
server, seeded, _ = stack
|
||||
with authenticated(server) as client:
|
||||
responses = [
|
||||
client.post("/api/v1/upload-preflight", json={}),
|
||||
client.get("/api/v1/workflow"),
|
||||
client.get("/api/v1/upload-batches"),
|
||||
client.get("/api/v1/assets/unknown-asset/thumbnail?size=256"),
|
||||
client.get("/api/v1/archive-plans/unknown-plan"),
|
||||
client.post("/api/v1/archive-locations", json={"name": "x", "root": "/nope"}),
|
||||
client.get("/api/v1/does-not-exist"),
|
||||
]
|
||||
for response in responses:
|
||||
assert SENTINEL_KEY not in response.text
|
||||
assert "Traceback" not in response.text
|
||||
if response.status_code >= 400:
|
||||
assert str(seeded.lib) not in response.text
|
||||
assert "photo_pipeline/" not in response.text
|
||||
assert "sqlite" not in response.text.lower()
|
||||
|
||||
|
||||
def test_a_symlink_swapped_under_an_asset_cannot_be_served(stack):
|
||||
"""TOCTOU on the media path: the file the database points at is replaced by a
|
||||
link to something outside the library between the scan and the request."""
|
||||
server, seeded, outside = stack
|
||||
secret = outside / "secret.jpg"
|
||||
original = seeded.lib / "b.jpg"
|
||||
original.unlink()
|
||||
original.symlink_to(secret)
|
||||
|
||||
with authenticated(server) as client:
|
||||
response = client.get(f"/api/v1/assets/{seeded.asset_ids['b']}/thumbnail?size=256")
|
||||
assert response.status_code == 403
|
||||
assert response.json()["error"]["code"] == "path_not_allowed"
|
||||
# The refusal names no filesystem location, and no bytes escaped with it.
|
||||
assert str(outside) not in response.text and str(seeded.lib) not in response.text
|
||||
assert secret.read_bytes()[:16] not in response.content
|
||||
|
||||
|
||||
def test_the_frontend_shell_stays_reachable_without_a_session(stack):
|
||||
"""It must load before any JavaScript can ask for a session."""
|
||||
server, _, _ = stack
|
||||
with anonymous(server) as client:
|
||||
response = client.get("/app/", headers={"Sec-Fetch-Site": "none"})
|
||||
assert response.status_code == 200
|
||||
assert response.headers["content-type"].startswith("text/html")
|
||||
assert Path("frontend/index.html").exists()
|
||||
325
tests/fixtures/media_corpus.py
vendored
325
tests/fixtures/media_corpus.py
vendored
@@ -1,325 +0,0 @@
|
||||
"""The golden media corpus: every format, orientation, profile, damage, and
|
||||
metadata case this application claims to survive (US07-03, concept §18).
|
||||
|
||||
``CASES`` is the manifest and the authority. Each entry declares a stable logical
|
||||
id (never a path), how the file is generated, and what the pipeline must do with
|
||||
it — decode it, or refuse it with one precise error code. Tests parametrize over the
|
||||
manifest, so a case that is added here without an expectation, or an expectation
|
||||
that stops holding, fails the suite rather than quietly going untested.
|
||||
|
||||
Everything is generated, never committed: fixed pixel seeds, fixed EXIF strings, no
|
||||
clock, no network, no personal data. Regeneration is byte-stable, which
|
||||
``test_media_hardening.py`` proves by building the corpus twice and comparing
|
||||
checksums — a golden corpus that drifts is not golden.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import struct
|
||||
import zlib
|
||||
from dataclasses import dataclass, field
|
||||
from pathlib import Path
|
||||
from typing import Callable
|
||||
|
||||
import numpy as np
|
||||
from PIL import Image
|
||||
|
||||
# Error codes the pipeline may answer with; ``None`` means "must render".
|
||||
UNSUPPORTED = "unsupported_image"
|
||||
TOO_LARGE = "image_too_large"
|
||||
|
||||
# One fixed capture time for every metadata case: the corpus must not depend on when
|
||||
# it was generated.
|
||||
CAPTURE_TIME = "2019:07:14 10:30:00"
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class MediaCase:
|
||||
id: str
|
||||
filename: str
|
||||
kind: str # format | orientation | profile | damaged | metadata
|
||||
build: Callable[[Path], None]
|
||||
expect_error: str | None = None
|
||||
# Declared for the metadata cases: exiftool arguments applied after the pixels
|
||||
# are written, and the user fields that must survive every later stage.
|
||||
exif_args: tuple[str, ...] = ()
|
||||
preserved_fields: tuple[str, ...] = ()
|
||||
notes: str = ""
|
||||
tags: tuple[str, ...] = field(default_factory=tuple)
|
||||
|
||||
|
||||
# ── generators ────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def _pixels(width: int, height: int, seed: int, bands: int = 3) -> np.ndarray:
|
||||
return np.random.default_rng(seed).integers(0, 256, (height, width, bands), dtype=np.uint8)
|
||||
|
||||
|
||||
def _save(path: Path, image: Image.Image, **kwargs) -> None:
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
image.save(path, **kwargs)
|
||||
|
||||
|
||||
def _jpeg(width=320, height=240, seed=1, **kwargs):
|
||||
def build(path: Path) -> None:
|
||||
_save(path, Image.fromarray(_pixels(width, height, seed)), quality=90, **kwargs)
|
||||
|
||||
return build
|
||||
|
||||
|
||||
def _oriented_jpeg(orientation: int):
|
||||
def build(path: Path) -> None:
|
||||
image = Image.fromarray(_pixels(400, 200, 5)) # landscape source
|
||||
exif = image.getexif()
|
||||
exif[274] = orientation # 0x0112 Orientation
|
||||
_save(path, image, exif=exif, quality=90)
|
||||
|
||||
return build
|
||||
|
||||
|
||||
def _rotated_pixels(path: Path) -> None:
|
||||
"""The same scene rotated in the pixels instead of in a tag."""
|
||||
_save(path, Image.fromarray(_pixels(400, 200, 5)).transpose(Image.ROTATE_90), quality=90)
|
||||
|
||||
|
||||
def _png(alpha: bool = False, seed: int = 2):
|
||||
def build(path: Path) -> None:
|
||||
if alpha:
|
||||
_save(path, Image.fromarray(_pixels(120, 90, seed, bands=4), "RGBA"))
|
||||
else:
|
||||
_save(path, Image.fromarray(_pixels(120, 90, seed)))
|
||||
|
||||
return build
|
||||
|
||||
|
||||
def _webp(path: Path) -> None:
|
||||
_save(path, Image.fromarray(_pixels(150, 100, 3)), quality=80)
|
||||
|
||||
|
||||
def _tiff(path: Path) -> None:
|
||||
_save(path, Image.fromarray(_pixels(140, 110, 4)))
|
||||
|
||||
|
||||
def _grayscale(path: Path) -> None:
|
||||
_save(path, Image.fromarray(_pixels(100, 100, 6)).convert("L"), quality=90)
|
||||
|
||||
|
||||
def _cmyk(path: Path) -> None:
|
||||
_save(path, Image.fromarray(_pixels(100, 100, 7)).convert("CMYK"), quality=90)
|
||||
|
||||
|
||||
def _tiny(path: Path) -> None:
|
||||
_save(path, Image.fromarray(_pixels(1, 1, 8)))
|
||||
|
||||
|
||||
def _icc_tagged(path: Path) -> None:
|
||||
"""A profile-bearing image: the colour-managed decode path must run."""
|
||||
from PIL import ImageCms
|
||||
|
||||
profile = bytearray(ImageCms.ImageCmsProfile(ImageCms.createProfile("sRGB")).tobytes())
|
||||
# An ICC header carries its creation timestamp at bytes 24..35. Left alone, the
|
||||
# corpus would be a different corpus every time it is generated.
|
||||
profile[24:36] = b"\x00" * 12
|
||||
_save(path, Image.fromarray(_pixels(120, 80, 9)), icc_profile=bytes(profile), quality=90)
|
||||
|
||||
|
||||
def _broken_icc(path: Path) -> None:
|
||||
"""A profile that is not a profile: a picture is still a picture."""
|
||||
_save(path, Image.fromarray(_pixels(120, 80, 10)), icc_profile=b"not-a-profile", quality=90)
|
||||
|
||||
|
||||
def _wide_jpeg(path: Path) -> None:
|
||||
"""Large enough that decoding it at full resolution is visible in memory."""
|
||||
_save(path, Image.fromarray(_pixels(4000, 3000, 11)), quality=70)
|
||||
|
||||
|
||||
def _zero_byte(path: Path) -> None:
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
path.write_bytes(b"")
|
||||
|
||||
|
||||
def _truncated_jpeg(path: Path) -> None:
|
||||
image = Image.fromarray(_pixels(400, 300, 12))
|
||||
_save(path, image, quality=90)
|
||||
data = path.read_bytes()
|
||||
path.write_bytes(data[: len(data) // 2]) # header intact, pixels missing
|
||||
|
||||
|
||||
def _corrupt_png(path: Path) -> None:
|
||||
image = Image.fromarray(_pixels(120, 90, 13))
|
||||
_save(path, image)
|
||||
data = bytearray(path.read_bytes())
|
||||
data[40:80] = b"\x00" * 40 # shred the compressed stream, keep the header
|
||||
path.write_bytes(bytes(data))
|
||||
|
||||
|
||||
def _not_an_image(path: Path) -> None:
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
path.write_bytes(b"This is a text file that happens to be named .jpg\n")
|
||||
|
||||
|
||||
def _png_declaring(width: int, height: int):
|
||||
"""A tiny, structurally valid PNG whose header claims an enormous picture.
|
||||
|
||||
A few hundred bytes on disk, gigapixels on paper: the pipeline must refuse it
|
||||
from the declared dimensions, before a single pixel is allocated. Rewriting the
|
||||
IHDR of a real PNG (rather than hand-rolling a stub) keeps the file openable, so
|
||||
the refusal is proven to come from the size check and not from a parse failure.
|
||||
"""
|
||||
|
||||
def build(path: Path) -> None:
|
||||
import io
|
||||
|
||||
buffer = io.BytesIO()
|
||||
Image.fromarray(_pixels(4, 4, 15)).save(buffer, format="PNG")
|
||||
data = bytearray(buffer.getvalue())
|
||||
start = 8 + 4 # PNG signature, then the IHDR length field
|
||||
struct.pack_into(">II", data, start + 4, width, height)
|
||||
ihdr = bytes(data[start : start + 4 + 13])
|
||||
struct.pack_into(">I", data, start + 4 + 13, zlib.crc32(ihdr))
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
path.write_bytes(bytes(data))
|
||||
|
||||
return build
|
||||
|
||||
|
||||
CASES: tuple[MediaCase, ...] = (
|
||||
# ── formats ───────────────────────────────────────────────────────────────
|
||||
MediaCase("jpeg", "formats/plain.jpg", "format", _jpeg()),
|
||||
MediaCase("jpeg_uppercase_ext", "formats/UPPER.JPG", "format", _jpeg(seed=14)),
|
||||
MediaCase("png", "formats/plain.png", "format", _png()),
|
||||
MediaCase("png_alpha", "formats/alpha.png", "format", _png(alpha=True)),
|
||||
MediaCase("webp", "formats/plain.webp", "format", _webp),
|
||||
MediaCase("tiff", "formats/plain.tiff", "format", _tiff),
|
||||
# ── orientation ───────────────────────────────────────────────────────────
|
||||
*(
|
||||
MediaCase(
|
||||
f"orientation_{value}",
|
||||
f"orientation/exif_{value}.jpg",
|
||||
"orientation",
|
||||
_oriented_jpeg(value),
|
||||
notes="EXIF orientation must be applied before resizing",
|
||||
)
|
||||
for value in range(1, 9)
|
||||
),
|
||||
MediaCase("rotated_pixels", "orientation/rotated.jpg", "orientation", _rotated_pixels),
|
||||
# ── colour and size profiles ──────────────────────────────────────────────
|
||||
MediaCase("grayscale", "profiles/gray.jpg", "profile", _grayscale),
|
||||
MediaCase("cmyk", "profiles/cmyk.jpg", "profile", _cmyk),
|
||||
MediaCase("tiny", "profiles/tiny.png", "profile", _tiny),
|
||||
MediaCase("icc_tagged", "profiles/icc.jpg", "profile", _icc_tagged),
|
||||
MediaCase(
|
||||
"broken_icc",
|
||||
"profiles/broken-icc.jpg",
|
||||
"profile",
|
||||
_broken_icc,
|
||||
notes="an unusable ICC profile falls back to a plain conversion, never an error",
|
||||
),
|
||||
MediaCase(
|
||||
"large_jpeg",
|
||||
"profiles/large.jpg",
|
||||
"profile",
|
||||
_wide_jpeg,
|
||||
notes="12 megapixels: the decode must stay near the requested size",
|
||||
),
|
||||
# ── damaged and hostile inputs ────────────────────────────────────────────
|
||||
MediaCase("zero_byte", "damaged/empty.jpg", "damaged", _zero_byte, UNSUPPORTED),
|
||||
MediaCase("truncated_jpeg", "damaged/truncated.jpg", "damaged", _truncated_jpeg, UNSUPPORTED),
|
||||
MediaCase("corrupt_png", "damaged/corrupt.png", "damaged", _corrupt_png, UNSUPPORTED),
|
||||
MediaCase("text_as_jpeg", "damaged/text.jpg", "damaged", _not_an_image, UNSUPPORTED),
|
||||
MediaCase(
|
||||
"bomb_header",
|
||||
"damaged/bomb.png",
|
||||
"damaged",
|
||||
_png_declaring(60_000, 60_000),
|
||||
TOO_LARGE,
|
||||
notes="3.6 gigapixels declared in the header and nothing else",
|
||||
),
|
||||
MediaCase(
|
||||
"bomb_warning_band",
|
||||
"damaged/bomb-warning.png",
|
||||
"damaged",
|
||||
_png_declaring(10_000, 10_000),
|
||||
TOO_LARGE,
|
||||
notes="inside Pillow's warn-only band; the warning is promoted to a refusal",
|
||||
),
|
||||
# ── metadata ──────────────────────────────────────────────────────────────
|
||||
MediaCase("no_exif", "metadata/bare.jpg", "metadata", _jpeg(seed=20)),
|
||||
MediaCase(
|
||||
"user_exif",
|
||||
"metadata/user.jpg",
|
||||
"metadata",
|
||||
_jpeg(seed=21),
|
||||
exif_args=(
|
||||
"-Artist=Ada Lovelace",
|
||||
"-Copyright=(c) Ada",
|
||||
f"-DateTimeOriginal={CAPTURE_TIME}",
|
||||
"-GPSLatitude=48.137",
|
||||
"-GPSLatitudeRef=N",
|
||||
"-Rating=4",
|
||||
"-ImageDescription=A day out",
|
||||
),
|
||||
preserved_fields=(
|
||||
"EXIF:IFD0:Artist",
|
||||
"EXIF:IFD0:Copyright",
|
||||
"EXIF:ExifIFD:DateTimeOriginal",
|
||||
"EXIF:IFD0:ImageDescription",
|
||||
"XMP:XMP-xmp:Rating",
|
||||
),
|
||||
notes="user metadata that every stage must leave exactly as it found it",
|
||||
),
|
||||
MediaCase(
|
||||
"prior_safety_keyword",
|
||||
"metadata/prior-safety.jpg",
|
||||
"metadata",
|
||||
_jpeg(seed=22),
|
||||
exif_args=("-Keywords+=nsfw", "-Subject+=nsfw", "-Artist=Ada Lovelace"),
|
||||
preserved_fields=("EXIF:IFD0:Artist",),
|
||||
notes="a safety decision already written by an earlier run",
|
||||
),
|
||||
MediaCase(
|
||||
"prior_analysis_keywords",
|
||||
"metadata/prior-analysis.jpg",
|
||||
"metadata",
|
||||
_jpeg(seed=23),
|
||||
exif_args=("-Keywords+=beach", "-Keywords+=sunset", "-Subject+=beach", "-Subject+=sunset"),
|
||||
notes="analysis keywords from an earlier run; a safety write must not drop them",
|
||||
),
|
||||
MediaCase(
|
||||
"conflicting_safety_keywords",
|
||||
"metadata/conflicting.jpg",
|
||||
"metadata",
|
||||
_jpeg(seed=24),
|
||||
exif_args=("-Keywords+=sfw", "-Keywords+=nsfw", "-Subject+=sfw", "-Subject+=nsfw"),
|
||||
notes="both safety keywords at once: mutually exclusive means one must go",
|
||||
),
|
||||
MediaCase(
|
||||
"malformed_metadata",
|
||||
"metadata/malformed.jpg",
|
||||
"metadata",
|
||||
_jpeg(seed=25, exif=b"\x00\x01\x02not-a-valid-exif-block"),
|
||||
notes="a broken EXIF block must not stop the picture from being usable",
|
||||
),
|
||||
)
|
||||
|
||||
CASES_BY_ID = {case.id: case for case in CASES}
|
||||
|
||||
|
||||
def build_corpus(root: Path, *, ids: tuple[str, ...] | None = None) -> dict[str, Path]:
|
||||
"""Generate the corpus (or a named subset) under ``root``; return id → path."""
|
||||
import subprocess
|
||||
|
||||
built: dict[str, Path] = {}
|
||||
for case in CASES:
|
||||
if ids is not None and case.id not in ids:
|
||||
continue
|
||||
path = root / case.filename
|
||||
case.build(path)
|
||||
if case.exif_args:
|
||||
subprocess.run(
|
||||
["exiftool", "-m", "-overwrite_original", *case.exif_args, str(path)],
|
||||
capture_output=True,
|
||||
check=False,
|
||||
)
|
||||
built[case.id] = path
|
||||
return built
|
||||
@@ -1,297 +0,0 @@
|
||||
"""US07-03: EXIF checkpoints, asserted with before/after metadata snapshots.
|
||||
|
||||
Every test here reads the complete metadata of a real file through exiftool before
|
||||
the stage runs and again afterwards, then compares the two. That is the only way to
|
||||
prove the property the concept actually asks for: a stage owns a few fields and must
|
||||
leave literally everything else — dates, GPS, camera, artist, rating, other people's
|
||||
keywords — exactly as it found them.
|
||||
|
||||
The other half is divergence. When something outside the stage's ownership does move,
|
||||
the checkpoint must say so, refuse to call itself verified, and change nothing back:
|
||||
a silent repair is how a library quietly loses the user's metadata.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import shutil
|
||||
import subprocess
|
||||
import uuid
|
||||
from datetime import datetime, timezone
|
||||
from types import SimpleNamespace
|
||||
|
||||
import pytest
|
||||
|
||||
from photo_pipeline.config import Config
|
||||
from photo_pipeline.db import create_db_engine, create_session_factory, run_migrations
|
||||
from photo_pipeline.integrations import exiftool
|
||||
from photo_pipeline.models import Asset, ExifProjection
|
||||
from photo_pipeline.services import exif_checkpoint
|
||||
from photo_pipeline.services.analysis import AnalysisService
|
||||
from photo_pipeline.services.safety import SafetyService
|
||||
from tests.fixtures.media_corpus import CASES_BY_ID, build_corpus
|
||||
|
||||
pytestmark = pytest.mark.skipif(
|
||||
shutil.which("exiftool") is None, reason="exiftool not installed"
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def env(tmp_path):
|
||||
data = tmp_path / "data"
|
||||
data.mkdir()
|
||||
lib = tmp_path / "lib"
|
||||
files = build_corpus(
|
||||
lib,
|
||||
ids=(
|
||||
"user_exif",
|
||||
"prior_safety_keyword",
|
||||
"prior_analysis_keywords",
|
||||
"conflicting_safety_keywords",
|
||||
"malformed_metadata",
|
||||
"no_exif",
|
||||
),
|
||||
)
|
||||
config = Config.from_env(
|
||||
{"PHOTO_PIPELINE_DATA_DIR": str(data), "PHOTO_PIPELINE_LIBRARY_ROOTS": str(lib)}
|
||||
)
|
||||
run_migrations(config.database_url)
|
||||
engine = create_db_engine(config.database_url)
|
||||
sf = create_session_factory(engine)
|
||||
assets = {}
|
||||
with sf() as session:
|
||||
for case_id, path in files.items():
|
||||
asset = Asset(
|
||||
id=str(uuid.uuid4()),
|
||||
original_path=str(path),
|
||||
current_path=str(path),
|
||||
discovered_at=datetime.now(timezone.utc),
|
||||
hash_version=1,
|
||||
)
|
||||
session.add(asset)
|
||||
assets[case_id] = asset.id
|
||||
session.commit()
|
||||
yield SimpleNamespace(config=config, lib=lib, sf=sf, files=files, assets=assets)
|
||||
engine.dispose()
|
||||
|
||||
|
||||
class Provider:
|
||||
"""A vision provider whose tags are fixed, so the EXIF assertion is the test."""
|
||||
|
||||
def __init__(self, tags):
|
||||
self.tags = list(tags)
|
||||
|
||||
def analyze(self, path, *, album_hint):
|
||||
return {"description": "a photo", "tags": self.tags}
|
||||
|
||||
|
||||
def snapshot(path):
|
||||
return exiftool.read_all(str(path))
|
||||
|
||||
|
||||
def stage_fields(before, after):
|
||||
"""Everything that changed except this application's owned and volatile tags."""
|
||||
return exif_checkpoint.compare(before, after)
|
||||
|
||||
|
||||
# ── preservation ──────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_a_safety_decision_preserves_every_user_field(env):
|
||||
case = CASES_BY_ID["user_exif"]
|
||||
path = env.files["user_exif"]
|
||||
before = snapshot(path)
|
||||
assert before, "the fixture must actually carry user metadata"
|
||||
|
||||
result = SafetyService(env.sf).decide(env.assets["user_exif"], "nsfw")
|
||||
after = snapshot(path)
|
||||
|
||||
assert result["exif_verified"] is True
|
||||
assert stage_fields(before, after) == ()
|
||||
for field in case.preserved_fields:
|
||||
assert after[field] == before[field], field
|
||||
assert "nsfw" in exif_checkpoint.owned_values(after)
|
||||
assert "sfw" not in exif_checkpoint.owned_values(after)
|
||||
|
||||
|
||||
def test_analysis_keywords_are_additive_and_keep_the_safety_decision(env):
|
||||
"""The two stages share the Keywords field; the second must merge, not replace."""
|
||||
asset_id = env.assets["prior_analysis_keywords"]
|
||||
path = env.files["prior_analysis_keywords"]
|
||||
SafetyService(env.sf).decide(asset_id, "sfw")
|
||||
before = snapshot(path)
|
||||
|
||||
AnalysisService(
|
||||
env.sf, provider=Provider(["harbour", "boats"]), library_roots=(env.lib,)
|
||||
).run([asset_id])
|
||||
after = snapshot(path)
|
||||
|
||||
keywords = exif_checkpoint.owned_values(after)
|
||||
assert {"sfw", "beach", "sunset", "harbour", "boats"} <= keywords
|
||||
assert "nsfw" not in keywords
|
||||
assert stage_fields(before, after) == ()
|
||||
|
||||
|
||||
def test_flipping_a_safety_decision_removes_only_the_opposite_keyword(env):
|
||||
asset_id = env.assets["prior_analysis_keywords"]
|
||||
path = env.files["prior_analysis_keywords"]
|
||||
safety = SafetyService(env.sf)
|
||||
safety.decide(asset_id, "nsfw")
|
||||
before = snapshot(path)
|
||||
|
||||
safety.decide(asset_id, "sfw")
|
||||
after = snapshot(path)
|
||||
|
||||
keywords = exif_checkpoint.owned_values(after)
|
||||
assert "sfw" in keywords and "nsfw" not in keywords
|
||||
assert {"beach", "sunset"} <= keywords, "analysis keywords are not safety's to remove"
|
||||
assert stage_fields(before, after) == ()
|
||||
|
||||
|
||||
def test_conflicting_safety_keywords_are_resolved_to_one(env):
|
||||
asset_id = env.assets["conflicting_safety_keywords"]
|
||||
SafetyService(env.sf).decide(asset_id, "sfw")
|
||||
keywords = exif_checkpoint.owned_values(snapshot(env.files["conflicting_safety_keywords"]))
|
||||
assert keywords & {"sfw", "nsfw"} == {"sfw"}
|
||||
|
||||
|
||||
def test_malformed_metadata_does_not_block_the_checkpoint(env):
|
||||
result = SafetyService(env.sf).decide(env.assets["malformed_metadata"], "nsfw")
|
||||
assert result["exif_verified"] is True
|
||||
|
||||
|
||||
# ── verification and the refreshed hash ───────────────────────────────────────
|
||||
|
||||
|
||||
def test_a_verified_checkpoint_refreshes_the_recorded_bytes(env):
|
||||
"""exiftool rewrites the container, so the stored SHA-256 must be the new one —
|
||||
upload compares against exactly these bytes."""
|
||||
asset_id = env.assets["no_exif"]
|
||||
path = env.files["no_exif"]
|
||||
SafetyService(env.sf).decide(asset_id, "sfw")
|
||||
|
||||
with env.sf() as session:
|
||||
asset = session.get(Asset, asset_id)
|
||||
row = session.get(ExifProjection, (asset_id, "safety"))
|
||||
import hashlib
|
||||
|
||||
on_disk = hashlib.sha256(path.read_bytes()).hexdigest()
|
||||
assert asset.current_sha256 == on_disk
|
||||
assert row.state == "verified" and row.result_file_sha256 == on_disk
|
||||
assert json.loads(row.desired_json) == {"add": ["sfw"], "remove": ["nsfw"]}
|
||||
|
||||
|
||||
def test_the_projection_survives_a_restart(env):
|
||||
asset_id = env.assets["user_exif"]
|
||||
SafetyService(env.sf).decide(asset_id, "nsfw")
|
||||
|
||||
engine = create_db_engine(env.config.database_url) # a fresh connection, as a restart is
|
||||
try:
|
||||
with create_session_factory(engine)() as session:
|
||||
row = session.get(ExifProjection, (asset_id, "safety"))
|
||||
assert row.state == "verified" and row.verified_at is not None
|
||||
finally:
|
||||
engine.dispose()
|
||||
|
||||
|
||||
# ── divergence ────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_a_field_the_stage_does_not_own_changing_is_divergent(env, monkeypatch):
|
||||
"""Something rewrote the artist while the safety keyword was being written.
|
||||
|
||||
The stage must not call that verified, must record what moved, and must not put
|
||||
the old value back — the file is now a question for a human.
|
||||
"""
|
||||
asset_id = env.assets["user_exif"]
|
||||
path = env.files["user_exif"]
|
||||
real_apply = exiftool.apply_keywords
|
||||
|
||||
def sabotage(target, *, add=(), remove=()):
|
||||
ok = real_apply(target, add=add, remove=remove)
|
||||
subprocess.run(
|
||||
["exiftool", "-m", "-overwrite_original", "-Artist=Someone Else", str(target)],
|
||||
capture_output=True,
|
||||
check=False,
|
||||
)
|
||||
return ok
|
||||
|
||||
monkeypatch.setattr(exiftool, "apply_keywords", sabotage)
|
||||
result = SafetyService(env.sf).decide(asset_id, "nsfw")
|
||||
|
||||
assert result["exif_verified"] is False, "a divergent checkpoint is not verified"
|
||||
with env.sf() as session:
|
||||
row = session.get(ExifProjection, (asset_id, "safety"))
|
||||
asset = session.get(Asset, asset_id)
|
||||
assert row.state == "divergent"
|
||||
assert "EXIF:IFD0:Artist" in json.loads(row.divergent_fields)
|
||||
assert row.verified_at is None
|
||||
# Not repaired, and not silently accepted as the current verified bytes.
|
||||
assert snapshot(path)["EXIF:IFD0:Artist"] == "Someone Else"
|
||||
assert asset.current_sha256 is None
|
||||
|
||||
|
||||
def test_a_divergent_asset_shows_up_in_the_review_queue(env, monkeypatch):
|
||||
asset_id = env.assets["user_exif"]
|
||||
monkeypatch.setattr(
|
||||
exif_checkpoint,
|
||||
"run",
|
||||
lambda *args, **kwargs: exif_checkpoint.CheckpointResult(
|
||||
exif_checkpoint.DIVERGENT, changed_fields=("EXIF:IFD0:Artist",), sha256="abc"
|
||||
),
|
||||
)
|
||||
SafetyService(env.sf).decide(asset_id, "sfw")
|
||||
|
||||
rows = SafetyService(env.sf).review_queue()["items"]
|
||||
row = next(item for item in rows if item["asset_id"] == asset_id)
|
||||
assert row["exif_state"] == "divergent"
|
||||
assert row["exif_verified"] is False
|
||||
|
||||
|
||||
def test_a_write_that_does_not_take_is_a_failure_not_a_verification(env, monkeypatch):
|
||||
monkeypatch.setattr(exiftool, "apply_keywords", lambda *a, **k: False)
|
||||
result = SafetyService(env.sf).decide(env.assets["no_exif"], "sfw")
|
||||
assert result["exif_verified"] is False
|
||||
assert exif_checkpoint.state_for(env.sf, env.assets["no_exif"], "safety") == "failed"
|
||||
|
||||
|
||||
def test_unreadable_metadata_is_a_failure_not_an_empty_snapshot(env, monkeypatch):
|
||||
"""``None`` from exiftool means "cannot answer"; treating it as "nothing there"
|
||||
would make every field look preserved."""
|
||||
monkeypatch.setattr(exiftool, "read_all", lambda path: None)
|
||||
result = exif_checkpoint.run(str(env.files["no_exif"]), add=("sfw",))
|
||||
assert result.state == "failed" and result.reason == "metadata_unreadable"
|
||||
|
||||
|
||||
def test_analysis_records_its_own_projection_separately(env):
|
||||
asset_id = env.assets["user_exif"]
|
||||
SafetyService(env.sf).decide(asset_id, "sfw")
|
||||
AnalysisService(env.sf, provider=Provider(["pier"]), library_roots=(env.lib,)).run([asset_id])
|
||||
|
||||
assert exif_checkpoint.state_for(env.sf, asset_id, "safety") == "verified"
|
||||
assert exif_checkpoint.state_for(env.sf, asset_id, "analysis") == "verified"
|
||||
|
||||
|
||||
# ── the comparison rules themselves ───────────────────────────────────────────
|
||||
|
||||
|
||||
def test_compare_ignores_owned_and_volatile_fields_only():
|
||||
before = {
|
||||
"EXIF:IFD0:Artist": "Ada",
|
||||
"IPTC:Keywords": ["holiday"],
|
||||
"File:System:FileSize": "3.8 kB",
|
||||
"File:CurrentIPTCDigest": "aaa",
|
||||
"XMP:XMP-x:XMPToolkit": "old",
|
||||
}
|
||||
after = {
|
||||
"EXIF:IFD0:Artist": "Ada",
|
||||
"IPTC:Keywords": ["holiday", "sfw"],
|
||||
"File:System:FileSize": "3.9 kB",
|
||||
"File:CurrentIPTCDigest": "bbb",
|
||||
"XMP:XMP-x:XMPToolkit": "new",
|
||||
}
|
||||
assert exif_checkpoint.compare(before, after) == ()
|
||||
|
||||
after["EXIF:GPS:GPSLatitude"] = "48.1" # an addition counts as much as a loss
|
||||
del after["EXIF:IFD0:Artist"]
|
||||
assert exif_checkpoint.compare(before, after) == ("EXIF:GPS:GPSLatitude", "EXIF:IFD0:Artist")
|
||||
@@ -1,238 +0,0 @@
|
||||
"""Importing the archived CLI's CSV state into the database (US07-01).
|
||||
|
||||
The donor's ``nsfw_scores.csv`` was keyed by path; the database is keyed by a
|
||||
stable asset id. Every case here is about that gap: a path that moved, a path that
|
||||
matches nothing, a score a human has already overruled, and a file that is simply
|
||||
malformed. The import may add evidence and must never invent an asset, overwrite a
|
||||
judgement, or fail silently — whatever it does ends up in the report.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import uuid
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
from sqlalchemy import select
|
||||
|
||||
from photo_pipeline.db import create_db_engine, create_session_factory, run_migrations
|
||||
from photo_pipeline.models import Asset, AssetPath, SafetyReview
|
||||
from photo_pipeline.services.legacy_import import (
|
||||
REVIEWER,
|
||||
LegacyImportService,
|
||||
write_report,
|
||||
)
|
||||
|
||||
NOW = datetime(2026, 1, 1, tzinfo=timezone.utc)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def factory(tmp_path):
|
||||
url = f"sqlite:///{tmp_path / 'legacy.db'}"
|
||||
run_migrations(url)
|
||||
engine = create_db_engine(url)
|
||||
yield create_session_factory(engine)
|
||||
engine.dispose()
|
||||
|
||||
|
||||
def _asset(sf, path: Path, *, previous: Path | None = None) -> str:
|
||||
asset_id = str(uuid.uuid4())
|
||||
with sf() as session:
|
||||
session.add(
|
||||
Asset(
|
||||
id=asset_id,
|
||||
original_path=str(previous or path),
|
||||
current_path=str(path),
|
||||
discovered_at=NOW,
|
||||
hash_version=1,
|
||||
)
|
||||
)
|
||||
session.add(AssetPath(asset_id=asset_id, path=str(path), valid_from=NOW))
|
||||
if previous is not None:
|
||||
session.add(
|
||||
AssetPath(
|
||||
asset_id=asset_id, path=str(previous), valid_from=NOW, valid_until=NOW
|
||||
)
|
||||
)
|
||||
session.commit()
|
||||
return asset_id
|
||||
|
||||
|
||||
def _review(sf, asset_id: str, *, score=None, decision=None, at=NOW) -> None:
|
||||
with sf() as session:
|
||||
session.add(
|
||||
SafetyReview(
|
||||
id=str(uuid.uuid4()),
|
||||
asset_id=asset_id,
|
||||
score=score,
|
||||
decision=decision,
|
||||
reviewer="dom",
|
||||
created_at=at,
|
||||
)
|
||||
)
|
||||
session.commit()
|
||||
|
||||
|
||||
def _csv(tmp_path, rows: str) -> Path:
|
||||
path = tmp_path / "nsfw_scores.csv"
|
||||
path.write_text("path,nsfw_score\n" + rows, encoding="utf-8")
|
||||
return path
|
||||
|
||||
|
||||
def _scores(sf) -> dict[str, float]:
|
||||
with sf() as session:
|
||||
return {
|
||||
review.asset_id: review.score
|
||||
for review in session.scalars(
|
||||
select(SafetyReview).order_by(SafetyReview.created_at)
|
||||
)
|
||||
}
|
||||
|
||||
|
||||
def test_scores_are_imported_onto_asset_identity(tmp_path, factory):
|
||||
a = _asset(factory, tmp_path / "a.jpg")
|
||||
# This one was scored under its old path and has since moved.
|
||||
b = _asset(factory, tmp_path / "moved" / "b.jpg", previous=tmp_path / "b.jpg")
|
||||
csv_path = _csv(
|
||||
tmp_path, f"{tmp_path / 'a.jpg'},0.9123\n{tmp_path / 'b.jpg'},0.0100\n"
|
||||
)
|
||||
|
||||
report = LegacyImportService(factory).import_nsfw_scores(csv_path)
|
||||
|
||||
assert report.counts == {
|
||||
"rows": 2,
|
||||
"imported": 2,
|
||||
"skipped_existing": 0,
|
||||
"unchanged": 0,
|
||||
"reviewed": 0,
|
||||
"unmatched": 0,
|
||||
"rejected": 0,
|
||||
"conflicts": 0,
|
||||
}
|
||||
assert _scores(factory) == {a: 0.9123, b: 0.0100}
|
||||
# The imported rows are evidence, not judgements: no decision is invented.
|
||||
with factory() as session:
|
||||
rows = list(session.scalars(select(SafetyReview)))
|
||||
assert {r.decision for r in rows} == {None}
|
||||
assert {r.reviewer for r in rows} == {REVIEWER}
|
||||
assert csv_path.exists(), "the CSV is left on disk as its own backup"
|
||||
|
||||
|
||||
def test_a_reviewed_asset_is_never_overwritten_by_the_csv(tmp_path, factory):
|
||||
asset_id = _asset(factory, tmp_path / "a.jpg")
|
||||
_review(factory, asset_id, score=0.2, decision="sfw")
|
||||
csv_path = _csv(tmp_path, f"{tmp_path / 'a.jpg'},0.9999\n")
|
||||
|
||||
report = LegacyImportService(factory).import_nsfw_scores(csv_path, overwrite=True)
|
||||
|
||||
assert (report.imported, report.reviewed) == (0, 1)
|
||||
assert report.conflicts[0]["decision"] == "sfw"
|
||||
assert report.conflicts[0]["csv"] == 0.9999
|
||||
with factory() as session:
|
||||
rows = list(session.scalars(select(SafetyReview)))
|
||||
assert len(rows) == 1 and rows[0].decision == "sfw"
|
||||
|
||||
|
||||
def test_an_unknown_path_is_reported_never_turned_into_an_asset(tmp_path, factory):
|
||||
_asset(factory, tmp_path / "a.jpg")
|
||||
csv_path = _csv(
|
||||
tmp_path, f"{tmp_path / 'a.jpg'},0.5000\n{tmp_path / 'ghost.jpg'},0.5000\n"
|
||||
)
|
||||
|
||||
report = LegacyImportService(factory).import_nsfw_scores(csv_path)
|
||||
|
||||
assert report.unmatched == [str(tmp_path / "ghost.jpg")]
|
||||
with factory() as session:
|
||||
assert session.scalar(select(Asset).where(Asset.current_path.like("%ghost%"))) is None
|
||||
|
||||
|
||||
def test_malformed_rows_are_rejected_with_their_reason(tmp_path, factory):
|
||||
_asset(factory, tmp_path / "a.jpg")
|
||||
csv_path = _csv(
|
||||
tmp_path,
|
||||
f"{tmp_path / 'a.jpg'},0.5000\n"
|
||||
f"{tmp_path / 'a.jpg'},0.5000\n" # the same path twice
|
||||
f"{tmp_path / 'b.jpg'},not-a-number\n"
|
||||
f"{tmp_path / 'c.jpg'},7.5\n" # outside 0..1: corruption, not a score
|
||||
",0.5\n", # no path at all
|
||||
)
|
||||
|
||||
report = LegacyImportService(factory).import_nsfw_scores(csv_path)
|
||||
|
||||
reasons = sorted(r["reason"] for r in report.rejected)
|
||||
assert reasons == ["duplicate_path", "missing_path", "unparsable_score", "unparsable_score"]
|
||||
assert report.imported == 1
|
||||
|
||||
|
||||
def test_rerunning_changes_nothing_and_a_differing_score_needs_overwrite(tmp_path, factory):
|
||||
asset_id = _asset(factory, tmp_path / "a.jpg")
|
||||
csv_path = _csv(tmp_path, f"{tmp_path / 'a.jpg'},0.5000\n")
|
||||
service = LegacyImportService(factory)
|
||||
service.import_nsfw_scores(csv_path)
|
||||
|
||||
again = service.import_nsfw_scores(csv_path)
|
||||
assert (again.imported, again.unchanged) == (0, 1)
|
||||
assert len(_scores(factory)) == 1
|
||||
|
||||
changed = _csv(tmp_path, f"{tmp_path / 'a.jpg'},0.8000\n")
|
||||
refused = service.import_nsfw_scores(changed)
|
||||
assert (refused.imported, refused.skipped_existing) == (0, 1)
|
||||
assert refused.conflicts[0]["current"] == 0.5
|
||||
|
||||
forced = service.import_nsfw_scores(changed, overwrite=True)
|
||||
assert forced.imported == 1
|
||||
with factory() as session:
|
||||
latest = list(session.scalars(select(SafetyReview).order_by(SafetyReview.created_at)))[-1]
|
||||
assert (latest.asset_id, latest.score) == (asset_id, 0.8)
|
||||
|
||||
|
||||
def test_a_dry_run_reports_without_writing(tmp_path, factory):
|
||||
_asset(factory, tmp_path / "a.jpg")
|
||||
csv_path = _csv(tmp_path, f"{tmp_path / 'a.jpg'},0.5000\n")
|
||||
|
||||
report = LegacyImportService(factory).import_nsfw_scores(csv_path, dry_run=True)
|
||||
|
||||
assert report.imported == 1
|
||||
assert _scores(factory) == {}, "a dry run must leave the database alone"
|
||||
|
||||
|
||||
def test_a_missing_csv_is_a_reported_outcome_not_a_crash(tmp_path, factory):
|
||||
report = LegacyImportService(factory).import_nsfw_scores(tmp_path / "nothing.csv")
|
||||
assert report.rejected == [{"path": str(tmp_path / "nothing.csv"), "reason": "csv_missing"}]
|
||||
assert report.rows == 0
|
||||
|
||||
|
||||
def test_the_report_is_written_where_it_can_be_audited(tmp_path, factory):
|
||||
_asset(factory, tmp_path / "a.jpg")
|
||||
csv_path = _csv(tmp_path, f"{tmp_path / 'a.jpg'},0.5000\n")
|
||||
report = LegacyImportService(factory).import_nsfw_scores(csv_path)
|
||||
|
||||
written = write_report(report, tmp_path / "data")
|
||||
|
||||
payload = json.loads(written.read_text(encoding="utf-8"))
|
||||
assert payload["imported"] == 1
|
||||
assert payload["source"] == str(csv_path)
|
||||
assert payload["schema_version"] == 1
|
||||
assert payload["generated_at"]
|
||||
|
||||
|
||||
def test_a_later_review_still_wins_after_an_import(tmp_path, factory):
|
||||
"""The import is a floor, not a ceiling: a human decision made afterwards is
|
||||
the current state, exactly as everywhere else in the app."""
|
||||
asset_id = _asset(factory, tmp_path / "a.jpg")
|
||||
csv_path = _csv(tmp_path, f"{tmp_path / 'a.jpg'},0.9000\n")
|
||||
LegacyImportService(factory).import_nsfw_scores(csv_path)
|
||||
# The import stamps itself with the wall clock, so "afterwards" has to be too.
|
||||
_review(
|
||||
factory,
|
||||
asset_id,
|
||||
score=0.9,
|
||||
decision="sfw",
|
||||
at=datetime.now(timezone.utc) + timedelta(hours=1),
|
||||
)
|
||||
|
||||
with factory() as session:
|
||||
latest = list(session.scalars(select(SafetyReview).order_by(SafetyReview.created_at)))[-1]
|
||||
assert latest.decision == "sfw"
|
||||
@@ -1,335 +0,0 @@
|
||||
"""US07-03: the golden media corpus, driven through the real decode path.
|
||||
|
||||
Every case in ``tests/fixtures/media_corpus.py`` is exercised here. The claims:
|
||||
|
||||
* a supported format/orientation/profile renders, with the orientation applied and
|
||||
the decode bounded to roughly the size that was asked for;
|
||||
* a damaged, empty, lying, or gigapixel file becomes one precise item error — the
|
||||
scan still finishes, the other assets still render, and a job that meets one keeps
|
||||
running;
|
||||
* a failure leaves no temporary file behind, and startup removes only the temporaries
|
||||
this service recognises;
|
||||
* cache invalidation follows the pixels, not the metadata.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import shutil
|
||||
import time
|
||||
import tracemalloc
|
||||
from types import SimpleNamespace
|
||||
|
||||
import pytest
|
||||
from PIL import Image
|
||||
|
||||
from photo_pipeline import imaging
|
||||
from photo_pipeline.config import Config
|
||||
from photo_pipeline.db import create_db_engine, create_session_factory, run_migrations
|
||||
from photo_pipeline.models import Asset, Thumbnail
|
||||
from photo_pipeline.services.inventory import InventoryService
|
||||
from photo_pipeline.services.thumbnails import (
|
||||
TEMP_SUFFIX,
|
||||
ThumbnailError,
|
||||
ThumbnailService,
|
||||
)
|
||||
from tests.fixtures.media_corpus import CASES, CASES_BY_ID, build_corpus
|
||||
|
||||
RENDERABLE = [case for case in CASES if case.expect_error is None]
|
||||
DAMAGED = [case for case in CASES if case.expect_error is not None]
|
||||
|
||||
needs_exiftool = pytest.mark.skipif(
|
||||
shutil.which("exiftool") is None, reason="exiftool not installed"
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def corpus(tmp_path_factory):
|
||||
root = tmp_path_factory.mktemp("corpus")
|
||||
return SimpleNamespace(root=root, files=build_corpus(root))
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def env(tmp_path, corpus):
|
||||
"""A library holding the whole corpus, scanned into a fresh database."""
|
||||
data = tmp_path / "data"
|
||||
data.mkdir()
|
||||
lib = tmp_path / "lib"
|
||||
shutil.copytree(corpus.root, lib)
|
||||
config = Config.from_env(
|
||||
{"PHOTO_PIPELINE_DATA_DIR": str(data), "PHOTO_PIPELINE_LIBRARY_ROOTS": str(lib)}
|
||||
)
|
||||
run_migrations(config.database_url)
|
||||
engine = create_db_engine(config.database_url)
|
||||
sf = create_session_factory(engine)
|
||||
scan = InventoryService(sf).scan(lib)
|
||||
by_id = {
|
||||
case.id: next(
|
||||
(aid for path, aid in scan.asset_ids.items() if path.endswith(case.filename)), None
|
||||
)
|
||||
for case in CASES
|
||||
}
|
||||
yield SimpleNamespace(
|
||||
config=config, lib=lib, sf=sf, scan=scan, assets=by_id,
|
||||
thumbs=ThumbnailService(sf, config),
|
||||
)
|
||||
engine.dispose()
|
||||
|
||||
|
||||
# ── the manifest itself ───────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_the_manifest_is_internally_consistent():
|
||||
assert len({case.id for case in CASES}) == len(CASES), "logical ids must be unique"
|
||||
assert len({case.filename for case in CASES}) == len(CASES), "paths must be unique"
|
||||
for case in CASES:
|
||||
assert case.kind in {"format", "orientation", "profile", "damaged", "metadata"}
|
||||
assert case.expect_error in (None, "unsupported_image", "image_too_large")
|
||||
|
||||
|
||||
def test_the_corpus_regenerates_byte_for_byte(tmp_path):
|
||||
"""A golden corpus that drifts between runs cannot be a golden corpus."""
|
||||
first = build_corpus(tmp_path / "one")
|
||||
second = build_corpus(tmp_path / "two")
|
||||
digests = {
|
||||
case_id: (
|
||||
hashlib.sha256(first[case_id].read_bytes()).hexdigest(),
|
||||
hashlib.sha256(second[case_id].read_bytes()).hexdigest(),
|
||||
)
|
||||
for case_id in first
|
||||
}
|
||||
drifted = [case_id for case_id, (a, b) in digests.items() if a != b]
|
||||
assert not drifted, f"non-deterministic fixtures: {drifted}"
|
||||
|
||||
|
||||
# ── bounded decode ────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
@pytest.mark.parametrize("case", RENDERABLE, ids=lambda case: case.id)
|
||||
def test_every_supported_case_renders(case, env):
|
||||
asset_id = env.assets[case.id]
|
||||
assert asset_id, f"{case.id} was not discovered by the scan"
|
||||
path = env.thumbs.generate(asset_id, 256)
|
||||
with Image.open(path) as thumb:
|
||||
assert thumb.format == "WEBP"
|
||||
assert max(thumb.size) <= 256
|
||||
assert min(thumb.size) >= 1
|
||||
|
||||
|
||||
@pytest.mark.parametrize("orientation", range(1, 9))
|
||||
def test_exif_orientation_is_applied_before_resizing(orientation, env):
|
||||
"""All eight tags: the 90° ones must come out portrait from a landscape source."""
|
||||
asset_id = env.assets[f"orientation_{orientation}"]
|
||||
with Image.open(env.thumbs.generate(asset_id, 256)) as thumb:
|
||||
rotated = orientation in (5, 6, 7, 8)
|
||||
assert (thumb.height > thumb.width) is rotated
|
||||
|
||||
|
||||
def test_transparency_and_grayscale_survive_the_pipeline(env):
|
||||
with Image.open(env.thumbs.generate(env.assets["png_alpha"], 256)) as thumb:
|
||||
assert "A" in thumb.getbands()
|
||||
with Image.open(env.thumbs.generate(env.assets["grayscale"], 256)) as thumb:
|
||||
assert thumb.size == (100, 100) # smaller than the request: never upscaled
|
||||
|
||||
|
||||
def test_a_broken_colour_profile_still_produces_a_picture(env):
|
||||
"""An unusable ICC profile is a metadata problem, not a reason to lose the preview."""
|
||||
for case_id in ("icc_tagged", "broken_icc"):
|
||||
with Image.open(env.thumbs.generate(env.assets[case_id], 256)) as thumb:
|
||||
assert thumb.size[0] > 0
|
||||
|
||||
|
||||
def test_a_large_jpeg_is_not_decoded_at_full_resolution(env):
|
||||
"""12 megapixels would be ~36 MB of pixels; the draft decode keeps it far below."""
|
||||
tracemalloc.start()
|
||||
try:
|
||||
env.thumbs.generate(env.assets["large_jpeg"], 256)
|
||||
_, peak = tracemalloc.get_traced_memory()
|
||||
finally:
|
||||
tracemalloc.stop()
|
||||
assert peak < 12_000_000, f"decode peaked at {peak} bytes"
|
||||
|
||||
|
||||
# ── damaged inputs ────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
@pytest.mark.parametrize("case", DAMAGED, ids=lambda case: case.id)
|
||||
def test_every_damaged_case_is_a_precise_item_error(case, env):
|
||||
asset_id = env.assets[case.id]
|
||||
assert asset_id, f"{case.id} was not discovered by the scan"
|
||||
with pytest.raises(ThumbnailError) as raised:
|
||||
env.thumbs.generate(asset_id, 256)
|
||||
assert raised.value.code == case.expect_error
|
||||
# Persisted, so a broken original is not re-decoded on every request...
|
||||
with env.sf() as session:
|
||||
rows = [
|
||||
row
|
||||
for row in session.query(Thumbnail).all()
|
||||
if row.asset_id == asset_id and row.state == "error"
|
||||
]
|
||||
assert rows and rows[0].error_code == case.expect_error
|
||||
# ...and the cached failure is the same precise error, not a generic one.
|
||||
with pytest.raises(ThumbnailError) as again:
|
||||
env.thumbs.generate(asset_id, 256)
|
||||
assert again.value.code == case.expect_error
|
||||
|
||||
|
||||
def test_a_gigapixel_header_is_refused_quickly_and_cheaply(env):
|
||||
"""The refusal must come from the declared size, not from decoding it."""
|
||||
tracemalloc.start()
|
||||
started = time.monotonic()
|
||||
try:
|
||||
with pytest.raises(ThumbnailError):
|
||||
env.thumbs.generate(env.assets["bomb_header"], 1280)
|
||||
_, peak = tracemalloc.get_traced_memory()
|
||||
finally:
|
||||
tracemalloc.stop()
|
||||
assert time.monotonic() - started < 5
|
||||
assert peak < 5_000_000, f"a 3.6 gigapixel header allocated {peak} bytes"
|
||||
|
||||
|
||||
def test_one_broken_file_stops_neither_the_scan_nor_its_neighbours(env):
|
||||
"""The whole corpus is in the library, damaged files included."""
|
||||
assert len(env.scan.asset_ids) >= len(CASES) - 1 # the empty file has no pixels to hash
|
||||
for case in RENDERABLE[:5]:
|
||||
assert env.thumbs.generate(env.assets[case.id], 256).exists()
|
||||
|
||||
|
||||
def test_a_job_that_meets_a_broken_file_gets_evidence_not_an_exception(env):
|
||||
"""``ensure_protected`` is what the archive lane calls; a plan must not die on
|
||||
one unreadable original."""
|
||||
evidence = env.thumbs.ensure_protected(env.assets["corrupt_png"])
|
||||
assert evidence["state"] == "unsupported"
|
||||
assert evidence["error_code"] == "unsupported_image"
|
||||
assert env.thumbs.ensure_protected(env.assets["jpeg"])["state"] == "ready"
|
||||
|
||||
|
||||
def test_undecodable_files_do_not_break_hashing(env):
|
||||
"""Discovery records what it can: byte identity always, pixel identity when the
|
||||
file has pixels."""
|
||||
with env.sf() as session:
|
||||
assets = {a.current_path: a for a in session.query(Asset).all()}
|
||||
for case in DAMAGED:
|
||||
asset = next((a for path, a in assets.items() if path.endswith(case.filename)), None)
|
||||
if asset is None:
|
||||
continue
|
||||
assert asset.current_sha256, "byte identity is always available"
|
||||
assert asset.pixel_sha256 is None, "undecodable files must not invent pixel identity"
|
||||
|
||||
|
||||
# ── cache lifecycle ───────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_a_failed_render_leaves_no_temporary_behind(env):
|
||||
with pytest.raises(ThumbnailError):
|
||||
env.thumbs.generate(env.assets["truncated_jpeg"], 256)
|
||||
leftovers = list(env.config.thumbnail_cache_dir.rglob(f"*{TEMP_SUFFIX}"))
|
||||
assert leftovers == []
|
||||
|
||||
|
||||
def test_startup_cleanup_removes_only_recognised_temporaries(env):
|
||||
cache = env.config.thumbnail_cache_dir
|
||||
env.thumbs.generate(env.assets["jpeg"], 256) # creates the cache directory
|
||||
stale = cache / f".abc123{TEMP_SUFFIX}"
|
||||
stale.write_bytes(b"half a thumbnail")
|
||||
innocent = cache / "keep-me.webp"
|
||||
innocent.write_bytes(b"not a temporary")
|
||||
|
||||
assert env.thumbs.cleanup_temp_files() == 1
|
||||
assert not stale.exists()
|
||||
assert innocent.exists()
|
||||
assert list(cache.rglob("*.webp")), "real cache entries survive"
|
||||
|
||||
|
||||
def test_metadata_only_change_reuses_the_thumbnail_and_a_pixel_change_does_not(env):
|
||||
"""The cache key follows the pixels: an EXIF edit must not cost a re-render."""
|
||||
asset_id = env.assets["jpeg"]
|
||||
first = env.thumbs.generate(asset_id, 256)
|
||||
|
||||
with env.sf() as session: # what a rescan records after an EXIF-only edit
|
||||
asset = session.get(Asset, asset_id)
|
||||
asset.current_sha256 = "different-bytes"
|
||||
session.commit()
|
||||
assert env.thumbs.generate(asset_id, 256) == first
|
||||
|
||||
with env.sf() as session: # a genuine pixel change
|
||||
asset = session.get(Asset, asset_id)
|
||||
asset.pixel_sha256 = "different-pixels"
|
||||
session.commit()
|
||||
assert env.thumbs.generate(asset_id, 256) != first
|
||||
|
||||
|
||||
def test_a_deleted_cache_file_is_regenerated(env):
|
||||
asset_id = env.assets["png"]
|
||||
path = env.thumbs.generate(asset_id, 256)
|
||||
path.unlink()
|
||||
regenerated = env.thumbs.generate(asset_id, 256)
|
||||
assert regenerated == path and regenerated.exists()
|
||||
|
||||
|
||||
# ── the imaging door itself ───────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_open_image_translates_every_decoder_failure(corpus):
|
||||
"""Callers must be able to catch two typed errors, never bare ``Exception``."""
|
||||
for case in DAMAGED:
|
||||
expected = imaging.ImageTooLarge if case.expect_error == "image_too_large" else imaging.UndecodableImage
|
||||
with pytest.raises(expected):
|
||||
with imaging.open_image(corpus.files[case.id]) as image:
|
||||
image.load()
|
||||
|
||||
|
||||
def test_open_image_refuses_more_pixels_than_the_caller_allowed(corpus):
|
||||
with pytest.raises(imaging.ImageTooLarge):
|
||||
with imaging.open_image(corpus.files["jpeg"], max_pixels=100):
|
||||
pass
|
||||
with imaging.open_image(corpus.files["jpeg"], max_pixels=100_000) as image:
|
||||
assert image.size == (320, 240)
|
||||
|
||||
|
||||
@needs_exiftool
|
||||
def test_malformed_metadata_does_not_stop_the_picture(env):
|
||||
"""A broken EXIF block is a metadata fact, not a decode failure."""
|
||||
assert env.thumbs.generate(env.assets["malformed_metadata"], 256).exists()
|
||||
with env.sf() as session:
|
||||
asset = session.get(Asset, env.assets["malformed_metadata"])
|
||||
assert asset.pixel_sha256, "pixels are still identifiable"
|
||||
|
||||
|
||||
def test_scoring_never_relaxes_truncated_image_handling(corpus):
|
||||
"""The donor's process-global ``LOAD_TRUNCATED_IMAGES`` is gone for good.
|
||||
|
||||
It is global state: switching it on for the safety model would also switch it on
|
||||
for hashing and preview rendering in the same process, and half a file would
|
||||
silently become a valid picture (donor_ledger: nt-score-model).
|
||||
"""
|
||||
import inspect
|
||||
|
||||
from PIL import ImageFile
|
||||
|
||||
from photo_pipeline.integrations import nsfw_model
|
||||
|
||||
code = [
|
||||
line
|
||||
for line in inspect.getsource(nsfw_model).splitlines()
|
||||
if not line.strip().startswith("#")
|
||||
]
|
||||
assert not any("LOAD_TRUNCATED_IMAGES" in line for line in code)
|
||||
|
||||
# And the door holds even when something else in the process turned it on —
|
||||
# the frozen donor does exactly that when the characterization suite imports it.
|
||||
previous = ImageFile.LOAD_TRUNCATED_IMAGES
|
||||
ImageFile.LOAD_TRUNCATED_IMAGES = True
|
||||
try:
|
||||
with pytest.raises(imaging.UndecodableImage):
|
||||
with imaging.open_image(corpus.files["truncated_jpeg"]) as image:
|
||||
image.load()
|
||||
assert ImageFile.LOAD_TRUNCATED_IMAGES is True, "the caller's setting is restored"
|
||||
finally:
|
||||
ImageFile.LOAD_TRUNCATED_IMAGES = previous
|
||||
|
||||
|
||||
def test_every_manifest_case_is_exercised():
|
||||
"""The corpus lint: no fixture may sit in the manifest untested."""
|
||||
covered = {case.id for case in RENDERABLE} | {case.id for case in DAMAGED}
|
||||
assert covered == set(CASES_BY_ID)
|
||||
@@ -3,8 +3,6 @@ same fixture files (real exiftool-written keywords)."""
|
||||
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
import numpy as np
|
||||
import pytest
|
||||
@@ -13,10 +11,6 @@ from PIL import Image
|
||||
from photo_pipeline.integrations import exiftool
|
||||
from photo_pipeline.services import safety
|
||||
|
||||
# The donor is frozen in the read-only archive (US07-01); parity still compares
|
||||
# against it, so this suite is the second and last place that may import it.
|
||||
ARCHIVED_SOURCES = Path(__file__).resolve().parents[2] / "legacy_cli_archive" / "src"
|
||||
|
||||
EXIFTOOL = shutil.which("exiftool")
|
||||
pytestmark = pytest.mark.skipif(EXIFTOOL is None, reason="exiftool not installed")
|
||||
|
||||
@@ -37,7 +31,6 @@ def _tag(path, keyword):
|
||||
|
||||
|
||||
def test_extracted_marks_and_partition_match_donor(tmp_path):
|
||||
sys.path.insert(0, str(ARCHIVED_SOURCES)) # the donor lives in the archive now
|
||||
donor_exif = pytest.importorskip("nsfwtag.exif")
|
||||
|
||||
nsfw = _jpeg(tmp_path / "nsfw.jpg", 1)
|
||||
|
||||
@@ -1,136 +0,0 @@
|
||||
"""US07-02: what an unhandled failure says, and where the library ends.
|
||||
|
||||
Two boundaries that only show up below the HTTP surface:
|
||||
|
||||
* Exception text is where internals leak — absolute paths, SQL, and occasionally a
|
||||
credential passed to the call that blew up. The first tests drive the real
|
||||
application with a route that raises such an exception (no production route does)
|
||||
and assert the client sees only a code.
|
||||
* The path the database recorded is not the path the filesystem will open a moment
|
||||
later. Analysis is the one stage whose bytes leave this machine, so it resolves
|
||||
the source against the library roots immediately before the provider call.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import uuid
|
||||
from datetime import datetime, timezone
|
||||
|
||||
import pytest
|
||||
from PIL import Image
|
||||
from starlette.testclient import TestClient
|
||||
|
||||
from photo_pipeline.api.app import create_app
|
||||
from photo_pipeline.config import Config
|
||||
from photo_pipeline.db import create_db_engine, create_session_factory, run_migrations
|
||||
from photo_pipeline.models import AnalysisResult, Asset
|
||||
from photo_pipeline.services.analysis import AnalysisService
|
||||
from photo_pipeline.services.safety import SafetyService
|
||||
|
||||
BOOM = "sqlite:///Users/someone/Pictures/private.db failed with key sk-secret-123"
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def client(tmp_path):
|
||||
app = create_app(Config(data_dir=tmp_path / "data"))
|
||||
|
||||
@app.get("/api/v1/boom")
|
||||
def boom():
|
||||
raise RuntimeError(BOOM)
|
||||
|
||||
with TestClient(app, raise_server_exceptions=False) as test_client:
|
||||
yield test_client
|
||||
|
||||
|
||||
def test_an_unhandled_error_returns_a_bare_envelope(client, caplog):
|
||||
with caplog.at_level(logging.ERROR):
|
||||
response = client.get("/api/v1/boom")
|
||||
|
||||
assert response.status_code == 500
|
||||
assert response.json() == {"error": {"code": "internal_error", "message": "internal error"}}
|
||||
assert BOOM not in response.text and "Traceback" not in response.text
|
||||
# The operator still gets the whole story, on the server side.
|
||||
assert BOOM in caplog.text
|
||||
|
||||
|
||||
def test_a_refusal_response_still_carries_the_default_headers(client):
|
||||
"""A 500 escaping the middleware's response path would also escape its headers."""
|
||||
response = client.get("/api/v1/boom")
|
||||
assert response.headers["x-content-type-options"] == "nosniff"
|
||||
assert response.headers["x-frame-options"] == "DENY"
|
||||
|
||||
|
||||
# ── the library boundary, revalidated at the moment of use ───────────────────
|
||||
|
||||
|
||||
class RecordingProvider:
|
||||
def __init__(self):
|
||||
self.calls = []
|
||||
|
||||
def analyze(self, path, *, album_hint):
|
||||
self.calls.append(path)
|
||||
return {"description": "a photo", "tags": []}
|
||||
|
||||
|
||||
def _library(tmp_path):
|
||||
lib = tmp_path / "lib"
|
||||
lib.mkdir()
|
||||
outside = tmp_path / "outside"
|
||||
outside.mkdir()
|
||||
for path in (lib / "inside.jpg", outside / "private.jpg"):
|
||||
Image.new("RGB", (8, 8), "blue").save(path)
|
||||
(tmp_path / "data").mkdir()
|
||||
config = Config.from_env({"PHOTO_PIPELINE_DATA_DIR": str(tmp_path / "data")})
|
||||
run_migrations(config.database_url)
|
||||
return lib, outside, create_session_factory(create_db_engine(config.database_url))
|
||||
|
||||
|
||||
def _sfw_asset(sf, path):
|
||||
asset = Asset(
|
||||
id=str(uuid.uuid4()),
|
||||
original_path=str(path),
|
||||
current_path=str(path),
|
||||
discovered_at=datetime.now(timezone.utc),
|
||||
hash_version=1,
|
||||
)
|
||||
with sf() as session:
|
||||
session.add(asset)
|
||||
session.commit()
|
||||
SafetyService(sf).decide(asset.id, "sfw", write_exif=False)
|
||||
return asset.id
|
||||
|
||||
|
||||
def test_analysis_will_not_send_a_file_that_left_the_library(tmp_path):
|
||||
"""A link swapped under an asset after the scan points at something the user
|
||||
never put in the library. Those bytes must not reach the vision provider — it is
|
||||
the one place in the pipeline where content leaves this machine."""
|
||||
lib, outside, sf = _library(tmp_path)
|
||||
inside = lib / "inside.jpg"
|
||||
asset_id = _sfw_asset(sf, inside)
|
||||
inside.unlink()
|
||||
inside.symlink_to(outside / "private.jpg")
|
||||
|
||||
provider = RecordingProvider()
|
||||
result = AnalysisService(sf, provider=provider, library_roots=(lib,)).run([asset_id])
|
||||
|
||||
assert provider.calls == [], "the provider must never have been constructed a request"
|
||||
assert result == {"analyzed": 0, "skipped": 0, "errors": 1}
|
||||
with sf() as session:
|
||||
row = session.get(AnalysisResult, asset_id)
|
||||
# The failure is visible and names no path.
|
||||
assert row.status == "error"
|
||||
assert "outside the configured library roots" in row.error_message
|
||||
assert str(outside) not in row.error_message
|
||||
|
||||
|
||||
def test_analysis_still_reads_a_file_that_stayed_inside(tmp_path):
|
||||
"""The guard must resolve real paths, not refuse everything."""
|
||||
lib, _, sf = _library(tmp_path)
|
||||
asset_id = _sfw_asset(sf, lib / "inside.jpg")
|
||||
|
||||
provider = RecordingProvider()
|
||||
result = AnalysisService(sf, provider=provider, library_roots=(lib,)).run([asset_id])
|
||||
|
||||
assert result["analyzed"] == 1
|
||||
assert provider.calls == [str((lib / "inside.jpg").resolve())]
|
||||
@@ -143,19 +143,6 @@
|
||||
],
|
||||
"US06-06": [
|
||||
"tests/e2e/test_phase_f_pipeline.py"
|
||||
],
|
||||
"US07-01": [
|
||||
"tests/unit/test_legacy_archive.py",
|
||||
"tests/integration/test_legacy_import.py"
|
||||
],
|
||||
"US07-02": [
|
||||
"tests/unit/test_security_policy.py",
|
||||
"tests/integration/test_security_boundaries.py",
|
||||
"tests/e2e/test_security.py"
|
||||
],
|
||||
"US07-03": [
|
||||
"tests/integration/test_media_hardening.py",
|
||||
"tests/integration/test_exif_checkpoints.py"
|
||||
]
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,173 +0,0 @@
|
||||
"""Archive lint (US07-01): the frozen CLI archive is complete, honest, and inert.
|
||||
|
||||
Freezing the donors is only worth something if three things stay true: the archive
|
||||
still holds exactly the bytes it claims, it holds no secret, and production cannot
|
||||
reach it. Each of those is a one-line mistake away — a helpful `sys.path` insert, a
|
||||
copied `.env`, an edited "just this once" source — so each is asserted here.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import re
|
||||
import subprocess
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
import yaml
|
||||
|
||||
REPO = Path(__file__).resolve().parents[2]
|
||||
ARCHIVE = REPO / "legacy_cli_archive"
|
||||
SOURCES = ARCHIVE / "src"
|
||||
LEDGER = ARCHIVE / "donor_ledger.yaml"
|
||||
CHECKSUMS = ARCHIVE / "CHECKSUMS.sha256"
|
||||
|
||||
# Everything the concept requires an archive to carry (§3 "Donor-first CLI
|
||||
# migration and archival"): the sources, their docs, a dependency lock, schema
|
||||
# notes, a redacted sample configuration, the ledger, and recorded checksums.
|
||||
REQUIRED_ARTIFACTS = (
|
||||
"README.md",
|
||||
"donor_ledger.yaml",
|
||||
"CHECKSUMS.sha256",
|
||||
"requirements-lock.txt",
|
||||
"photo_analyzer.env.sample",
|
||||
)
|
||||
ARCHIVED_MODULES = ("photo_analyzer", "nsfwtag", "webapp", "nsfw_tag", "compare_models")
|
||||
# Only the two suites that compare against the donors may put the archive on the
|
||||
# import path; every other tree must not name it at all.
|
||||
IMPORT_ALLOWED = {
|
||||
REPO / "tests" / "characterization" / "conftest.py",
|
||||
REPO / "tests" / "characterization" / "test_donor_ledger.py", # lints the ledger there
|
||||
REPO / "tests" / "integration" / "test_safety_parity.py",
|
||||
Path(__file__),
|
||||
}
|
||||
SECRET_PATTERNS = (
|
||||
re.compile(r"sk-(?!REPLACE_WITH_YOUR_KEY)[A-Za-z0-9_\-]{16,}"),
|
||||
re.compile(r"AIza[0-9A-Za-z_\-]{20,}"), # Google API keys
|
||||
re.compile(r"(?i)api[_-]?key\s*[=:]\s*['\"][A-Za-z0-9_\-]{16,}['\"]"),
|
||||
)
|
||||
|
||||
|
||||
def _archived_files() -> list[Path]:
|
||||
return sorted(p for p in SOURCES.rglob("*") if p.is_file() and "__pycache__" not in p.parts)
|
||||
|
||||
|
||||
# ── completeness ─────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_the_archive_carries_every_required_artifact():
|
||||
for artifact in REQUIRED_ARTIFACTS:
|
||||
path = ARCHIVE / artifact
|
||||
assert path.is_file(), f"archive is missing {artifact}"
|
||||
assert path.stat().st_size > 0, f"archive artifact {artifact} is empty"
|
||||
assert (SOURCES / "photo_analyzer.py").is_file()
|
||||
assert (SOURCES / "nsfwtag" / "README.md").is_file(), "donor docs must be archived too"
|
||||
assert (SOURCES / "webapp" / "README.md").is_file()
|
||||
|
||||
|
||||
def test_every_ledger_source_is_present_in_the_archive():
|
||||
rows = yaml.safe_load(LEDGER.read_text(encoding="utf-8"))["rows"]
|
||||
for row in rows:
|
||||
source = SOURCES / row["source"]["file"]
|
||||
assert source.is_file(), f"{row['id']}: {row['source']['file']} is not archived"
|
||||
|
||||
|
||||
def test_the_readme_records_provenance_and_the_no_import_rule():
|
||||
readme = (ARCHIVE / "README.md").read_text(encoding="utf-8")
|
||||
for expected in ("CHECKSUMS.sha256", "requirements-lock.txt", "donor_ledger.yaml"):
|
||||
assert expected in readme, f"README does not point at {expected}"
|
||||
assert "Schema notes" in readme, "the donor's schema must be documented"
|
||||
assert "nsfw_scores.csv" in readme, "the CSV's fate must be documented"
|
||||
|
||||
|
||||
# ── integrity ────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_every_archived_source_matches_its_checksum():
|
||||
"""A frozen archive that silently drifts is not evidence of anything."""
|
||||
recorded = {}
|
||||
for line in CHECKSUMS.read_text(encoding="utf-8").splitlines():
|
||||
digest, _, rel = line.partition(" ")
|
||||
if rel:
|
||||
recorded[rel.strip()] = digest
|
||||
actual = {
|
||||
str(path.relative_to(ARCHIVE)): hashlib.sha256(path.read_bytes()).hexdigest()
|
||||
for path in _archived_files()
|
||||
}
|
||||
assert actual == recorded, "archived sources and CHECKSUMS.sha256 disagree"
|
||||
|
||||
|
||||
# ── redaction ────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_the_archive_contains_no_credential():
|
||||
for path in [*_archived_files(), *(ARCHIVE / a for a in REQUIRED_ARTIFACTS)]:
|
||||
if path.suffix in (".png", ".jpg", ".webp"):
|
||||
continue
|
||||
text = path.read_text(encoding="utf-8", errors="replace")
|
||||
for pattern in SECRET_PATTERNS:
|
||||
assert not pattern.search(text), f"possible secret in {path.relative_to(REPO)}"
|
||||
|
||||
|
||||
def test_the_sample_configuration_is_a_placeholder_only():
|
||||
sample = (ARCHIVE / "photo_analyzer.env.sample").read_text(encoding="utf-8")
|
||||
assert "REDACTED" in sample
|
||||
# Placeholders are not key-shaped, so neither a scanner nor a reader can
|
||||
# mistake the sample for a credential.
|
||||
assert "LLM_API_KEY=<" in sample
|
||||
# No real env file, database, log, or CSV may ride along in the archive.
|
||||
strays = [
|
||||
p.name
|
||||
for p in ARCHIVE.rglob("*")
|
||||
if p.is_file() and p.suffix in (".env", ".db", ".sqlite", ".sqlite3", ".log", ".csv")
|
||||
]
|
||||
assert strays == [], f"unexpected runtime files archived: {strays}"
|
||||
|
||||
|
||||
# ── inertness ────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_production_code_never_imports_an_archived_module():
|
||||
for path in (REPO / "photo_pipeline").rglob("*.py"):
|
||||
text = path.read_text(encoding="utf-8")
|
||||
for module in ARCHIVED_MODULES:
|
||||
assert not re.search(rf"^\s*(import|from)\s+{module}\b", text, re.MULTILINE), \
|
||||
f"{path.relative_to(REPO)} imports the archived {module}"
|
||||
assert "legacy_cli_archive" not in text, \
|
||||
f"{path.relative_to(REPO)} names the archive"
|
||||
|
||||
|
||||
def test_only_the_parity_suites_put_the_archive_on_the_import_path():
|
||||
for path in (REPO / "tests").rglob("*.py"):
|
||||
if path in IMPORT_ALLOWED or "__pycache__" in path.parts:
|
||||
continue
|
||||
assert "legacy_cli_archive" not in path.read_text(encoding="utf-8"), \
|
||||
f"{path.relative_to(REPO)} reaches into the archive"
|
||||
|
||||
|
||||
def test_the_archived_modules_are_unimportable_from_a_clean_interpreter():
|
||||
"""The real check: a fresh process with the repo on its path cannot load them."""
|
||||
script = (
|
||||
"import importlib.util, sys; "
|
||||
f"sys.path.insert(0, {str(REPO)!r}); "
|
||||
"print([m for m in "
|
||||
f"{list(ARCHIVED_MODULES)!r}"
|
||||
" if importlib.util.find_spec(m) is not None])"
|
||||
)
|
||||
result = subprocess.run(
|
||||
[sys.executable, "-c", script], capture_output=True, text=True, cwd=str(REPO), check=True
|
||||
)
|
||||
assert result.stdout.strip() == "[]", f"still importable: {result.stdout.strip()}"
|
||||
|
||||
|
||||
def test_the_application_starts_without_the_archive(tmp_path):
|
||||
"""Nothing in the runtime path may need the frozen sources to exist."""
|
||||
pytest.importorskip("fastapi")
|
||||
from photo_pipeline.api.app import create_app
|
||||
from photo_pipeline.config import Config
|
||||
|
||||
config = Config.from_env(
|
||||
{"PHOTO_PIPELINE_DATA_DIR": str(tmp_path / "data"), "PHOTO_PIPELINE_LIBRARY_ROOTS": ""}
|
||||
)
|
||||
assert create_app(config) is not None
|
||||
@@ -81,42 +81,3 @@ def test_symlink_within_root_is_allowed(tmp_path):
|
||||
pytest.skip("cannot create symlink on this platform")
|
||||
found = path_policy.discover([root])
|
||||
assert target in found and link in found
|
||||
|
||||
|
||||
def test_resolve_in_roots_returns_the_path_the_caller_must_use(tmp_path):
|
||||
"""The resolved path is the answer, not a yes/no: a caller that revalidates one
|
||||
name and then opens another has an open symlink race (US07-02)."""
|
||||
root = tmp_path / "lib"
|
||||
(root / "sub").mkdir(parents=True)
|
||||
target = root / "sub" / "real.jpg"
|
||||
target.write_bytes(b"x")
|
||||
link = root / "alias.jpg"
|
||||
os.symlink(target, link)
|
||||
|
||||
assert path_policy.resolve_in_roots([root], link) == target.resolve()
|
||||
assert path_policy.resolve_in_roots([root], target) == target.resolve()
|
||||
|
||||
|
||||
def test_resolve_in_roots_refuses_escapes_without_naming_them(tmp_path):
|
||||
root = tmp_path / "lib"
|
||||
root.mkdir()
|
||||
outside = tmp_path / "outside.jpg"
|
||||
outside.write_bytes(b"x")
|
||||
link = root / "alias.jpg"
|
||||
os.symlink(outside, link)
|
||||
|
||||
for candidate in (link, outside, root / "_IGNORE" / "a.jpg"):
|
||||
with pytest.raises(path_policy.PathPolicyError) as raised:
|
||||
path_policy.resolve_in_roots([root], candidate)
|
||||
assert str(outside) not in str(raised.value)
|
||||
|
||||
|
||||
def test_resolve_in_roots_checks_every_configured_root(tmp_path):
|
||||
first, second = tmp_path / "one", tmp_path / "two"
|
||||
first.mkdir()
|
||||
second.mkdir()
|
||||
photo = second / "b.jpg"
|
||||
photo.write_bytes(b"x")
|
||||
assert path_policy.resolve_in_roots([first, second], photo) == photo.resolve()
|
||||
# No configured boundary means nothing to check against.
|
||||
assert path_policy.resolve_in_roots([], photo) == photo
|
||||
|
||||
@@ -1,169 +0,0 @@
|
||||
"""US07-02: the request-admission policy, enumerated.
|
||||
|
||||
``evaluate`` decides every refusal the API can make before a route runs, so the
|
||||
whole local-web threat model is one table here: who may call, from where, with what
|
||||
proof. The middleware and the endpoints are covered black box in
|
||||
``tests/e2e/test_security.py``; this file pins the rules themselves, including the
|
||||
combinations a browser can produce but a test client rarely does.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
|
||||
from photo_pipeline.api.security import (
|
||||
CSRF_HEADER,
|
||||
PUBLIC_PATHS,
|
||||
Session,
|
||||
evaluate,
|
||||
split_host,
|
||||
)
|
||||
|
||||
SESSION = Session(id="session-id", csrf_token="csrf-token")
|
||||
HOST = "127.0.0.1:8000"
|
||||
LIMIT = 1024
|
||||
|
||||
|
||||
def check(method="GET", path="/api/v1/workflow", **headers):
|
||||
"""Evaluate a request that is authenticated and same-origin unless overridden."""
|
||||
sent = {
|
||||
"host": HOST,
|
||||
"cookie-session": SESSION.id,
|
||||
CSRF_HEADER: SESSION.csrf_token,
|
||||
}
|
||||
sent.update({name.replace("_", "-"): value for name, value in headers.items()})
|
||||
sent = {name: value for name, value in sent.items() if value is not None}
|
||||
return evaluate(
|
||||
method=method,
|
||||
path=path,
|
||||
headers=sent,
|
||||
session=SESSION,
|
||||
max_request_bytes=LIMIT,
|
||||
)
|
||||
|
||||
|
||||
def test_an_authenticated_same_origin_request_is_admitted():
|
||||
assert check() is None
|
||||
assert check(method="POST", origin="http://127.0.0.1:8000") is None
|
||||
assert check(sec_fetch_site="same-origin") is None
|
||||
|
||||
|
||||
@pytest.mark.parametrize("host", ["evil.example", "evil.example:8000", "192.168.1.10:8000", ""])
|
||||
def test_a_non_loopback_host_is_refused(host):
|
||||
"""DNS rebinding: the browser thinks it is talking to the attacker's name, which
|
||||
resolves to 127.0.0.1. The name is the evidence, so the name is checked."""
|
||||
refusal = check(host=host)
|
||||
assert (refusal.status, refusal.code) == (403, "host_not_allowed")
|
||||
|
||||
|
||||
@pytest.mark.parametrize("host", ["127.0.0.1:8000", "localhost:8000", "[::1]:8000", "localhost"])
|
||||
def test_loopback_hosts_are_accepted(host):
|
||||
assert check(host=host) is None
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"origin",
|
||||
[
|
||||
"http://evil.example",
|
||||
"https://evil.example:8000",
|
||||
"http://127.0.0.1:9999", # another local app is still another origin
|
||||
"http://localhost.evil.example:8000",
|
||||
"null",
|
||||
"file://",
|
||||
],
|
||||
)
|
||||
def test_a_foreign_origin_is_refused(origin):
|
||||
refusal = check(method="POST", origin=origin)
|
||||
assert (refusal.status, refusal.code) == (403, "origin_not_allowed")
|
||||
|
||||
|
||||
@pytest.mark.parametrize("origin", ["http://127.0.0.1:8000", "http://localhost:8000"])
|
||||
def test_this_applications_origin_is_accepted(origin):
|
||||
assert check(method="POST", origin=origin) is None
|
||||
|
||||
|
||||
@pytest.mark.parametrize("site", ["cross-site", "same-site"])
|
||||
def test_a_cross_site_fetch_is_refused_even_without_an_origin(site):
|
||||
"""What ``<img src="http://127.0.0.1:8000/...">`` on another page looks like."""
|
||||
refusal = check(path="/api/v1/assets/a1/thumbnail", sec_fetch_site=site)
|
||||
assert (refusal.status, refusal.code) == (403, "cross_site_blocked")
|
||||
|
||||
|
||||
def test_a_user_initiated_navigation_is_accepted():
|
||||
assert check(sec_fetch_site="none") is None
|
||||
|
||||
|
||||
def test_a_request_without_a_session_is_unauthenticated():
|
||||
for method, path in [
|
||||
("GET", "/api/v1/workflow"),
|
||||
("POST", "/api/v1/jobs"),
|
||||
("GET", "/api/v1/assets/a1/thumbnail"),
|
||||
]:
|
||||
refusal = check(method=method, path=path, cookie_session=None)
|
||||
assert (refusal.status, refusal.code) == (401, "unauthenticated"), path
|
||||
|
||||
|
||||
def test_a_forged_session_is_unauthenticated():
|
||||
refusal = check(cookie_session="guessed")
|
||||
assert (refusal.status, refusal.code) == (401, "unauthenticated")
|
||||
|
||||
|
||||
@pytest.mark.parametrize("path", sorted(PUBLIC_PATHS))
|
||||
def test_health_and_the_bootstrap_stay_reachable_without_a_session(path):
|
||||
assert check(path=path, cookie_session=None) is None
|
||||
|
||||
|
||||
def test_the_static_shell_needs_no_session():
|
||||
"""It has to load before any JavaScript can ask for one."""
|
||||
assert check(path="/app/index.html", cookie_session=None) is None
|
||||
|
||||
|
||||
@pytest.mark.parametrize("method", ["POST", "PUT", "PATCH", "DELETE"])
|
||||
def test_a_mutation_without_a_valid_csrf_token_is_refused(method):
|
||||
for token in (None, "guessed"):
|
||||
refusal = check(method=method, path="/api/v1/jobs", **{CSRF_HEADER: token})
|
||||
assert (refusal.status, refusal.code) == (403, "csrf_failed")
|
||||
|
||||
|
||||
@pytest.mark.parametrize("method", ["GET", "HEAD", "OPTIONS"])
|
||||
def test_reads_need_no_csrf_token(method):
|
||||
assert check(method=method, **{CSRF_HEADER: None}) is None
|
||||
|
||||
|
||||
def test_an_oversized_body_is_refused_before_it_is_read():
|
||||
refusal = check(method="POST", path="/api/v1/jobs", content_length=str(LIMIT + 1))
|
||||
assert (refusal.status, refusal.code) == (413, "payload_too_large")
|
||||
assert check(method="POST", path="/api/v1/jobs", content_length=str(LIMIT)) is None
|
||||
|
||||
|
||||
def test_the_host_check_precedes_authentication():
|
||||
"""A refusal must not tell a foreign caller whether its session guess was right."""
|
||||
refusal = check(host="evil.example", cookie_session="guessed")
|
||||
assert refusal.code == "host_not_allowed"
|
||||
|
||||
|
||||
def test_refusals_name_no_path_secret_or_internal():
|
||||
refusals = [
|
||||
check(host="evil.example"),
|
||||
check(method="POST", origin="http://evil.example"),
|
||||
check(cookie_session=None),
|
||||
check(method="POST", **{CSRF_HEADER: None}),
|
||||
]
|
||||
for refusal in refusals:
|
||||
assert SESSION.id not in refusal.message
|
||||
assert SESSION.csrf_token not in refusal.message
|
||||
assert "/" not in refusal.message
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"value,expected",
|
||||
[
|
||||
("127.0.0.1:8000", ("127.0.0.1", "8000")),
|
||||
("localhost", ("localhost", "")),
|
||||
("[::1]:8000", ("[::1]", "8000")),
|
||||
("[::1]", ("[::1]", "")),
|
||||
("", ("", "")),
|
||||
],
|
||||
)
|
||||
def test_split_host(value, expected):
|
||||
assert split_host(value) == expected
|
||||
Reference in New Issue
Block a user