Stream path security parity: bring push/stream defenses to the level of bundle/presign

Problem

The presign/bundle path and the stream path are two entry points for the same operation (push), but only the bundle path received the full security treatment from issues #49 (now closed). A pusher who stays under the 500-object threshold — or deliberately crafts pushes to stay there — bypasses every control added in phases 1–4 of #49.

This is not hypothetical. The stream path is the future micropayment-capable path and will see deliberate use. It must be hardened to parity before it sees production traffic.

Gaps (confirmed by code audit)

Implementation plan

Phase 1 — Blocked-hash check (critical, implement first)

Invariant: no object whose object_id appears in musehub_blocked_hashes may be written to storage or the DB via the stream path.

Where: inside _flush_batch in wire_push_stream, after the bulk DB-presence check and before any MinIO PUT.

Behaviour:

• Bulk-query musehub_blocked_hashes for all incoming_ids in the batch (one extra query per flush, same pattern as bundle path).
• If any match: yield ERROR frame (403 body), record a musehub_push_anomaly row flagging the attempt, return without writing.
• If no match: continue as today.

Tests (tests/test_stream_security_phase1.py):

• Blocked object in a single-batch stream → ERROR frame, nothing written to DB or MinIO.
• Blocked object in a multi-batch stream (arrives in batch 2) → ERROR frame, partial batch 1 writes do NOT roll back (content-addressing makes partial writes safe; the commit is never advanced).
• Clean stream bypasses check without regression.
• Idempotent: same blocked object pushed twice yields ERROR both times.

Phase 2 — Daily byte limit (rate limiting parity)

Invariant: the per-user daily byte counter (musehub_daily_push_bytes) is incremented for pushes via the stream path, and the pre-check fires before the stream is accepted.

Where:

• Pre-check: at the start of wire_push_stream, after auth, before entering the frame loop. Read today's total from musehub_daily_push_bytes and yield ERROR (429-equivalent) if already at or over bundle_daily_upload_limit_bytes.
• Accounting: at the END of a successful push (after branch pointer is advanced), call record_bundle_bytes_uploaded with the total raw bytes received.

Note: checking at stream-open time rather than at a presign step means the limit is advisory for streams that start just under the threshold and push a lot of data. This is acceptable — streams are small by definition (< 500 objects / < 50 MB). An exact byte-level gate would require buffering the entire stream first, which defeats the point of streaming.

Tests (tests/test_stream_security_phase2.py):

• Stream push increments musehub_daily_push_bytes by the correct byte count.
• User at the daily limit is rejected with an ERROR frame at stream open (before any frame is read).
• User under the limit completes successfully.
• Per-user isolation: one user exhausted does not block another.

Phase 3 — Cumulative decompressed size gate (zip-bomb defense)

Invariant: the sum of len(raw_content) across all objects in a single stream may not exceed bundle_max_decompressed_bytes (currently 4 GB). Breaching it terminates the stream with an ERROR frame.

Where: running accumulator inside the frame loop, incremented after each successful decompression (same position as the bundle path's _total_decompressed counter).

Behaviour:

• On breach: yield ERROR frame (413), return. No quarantine record needed (stream abuse is stateless — the connection drops).
• The per-object wire cap (50 MB compressed) stays; this adds a cumulative cap on top.

Tests (tests/test_stream_security_phase3.py):

• Stream with one object whose decompressed size exceeds the cumulative cap alone → ERROR.
• Stream where no single object exceeds the cap but the total does → ERROR on the frame that crosses the threshold.
• Stream well under the cap → no regression.
• Cap is read from settings.bundle_max_decompressed_bytes (configurable, not hardcoded).

Phase 4 — content.scan job enqueue (content moderation parity)

Invariant: every object successfully written to storage via the stream path gets a content.scan background job, exactly as the bundle path does.

Where: after _flush_batch returns stored objects, session.add one MusehubBackgroundJob(job_type="content.scan", ...)\ per newly-stored object. Skipped (already-present) objects do not get a duplicate scan job.

Tests (tests/test_stream_security_phase4.py):

• N new objects pushed via stream → N content.scan jobs enqueued.
• M objects skipped (already present) → 0 additional scan jobs for skipped objects.
• Zero objects pushed (all skipped) → 0 scan jobs.

Phase 5 — Quarantine DB record on stream abuse (observability parity)

Invariant: when the stream path rejects a push for a security reason (blocked hash, byte limit, zip bomb), a structured record is written to the DB so operators can investigate. An ERROR frame is not enough — it disappears with the connection.

New table (or reuse musehub_push_anomaly): musehub_stream_rejection with columns:

• rejection_id (sha256 genesis)
• identity_id
• repo_id
• reason (enum: blocked_hash | daily_limit | zip_bomb | other)
• detail (text, e.g. the blocked object_id)
• rejected_at

Tests (tests/test_stream_security_phase5.py):

• Blocked-hash rejection → musehub_stream_rejection row with reason=blocked_hash.
• Daily-limit rejection → row with reason=daily_limit.
• Zip-bomb rejection → row with reason=zip_bomb.
• Rejection row is written even if the session has a partial transaction (no silent swallow).

Acceptance criteria (all phases)

• All existing stream-path tests continue to pass.
• Each phase ships with its own TDD test file; tests are written first, confirmed failing, then implemented to green.
• No phase may be merged if it breaks the push/stream happy path.
• After phase 1, a DMCA-blocked object cannot enter storage via any push path.
• After phase 2, the daily byte counter reflects both presign and stream pushes.
• After phase 3, no single stream can decompress more than bundle_max_decompressed_bytes total.

Order

Phases 1 → 2 → 3 → 4 → 5. Phase 1 is the critical one — ship it first.