test(review): Phase 0 safety net for round-2/3 fixes

Host-testable guards before the perf/refactor phases touch behavior:
- test_download.py: _reject_rebound_internal (SSRF DNS-rebinding) — literal
  internal/public IPs, all-internal vs mixed resolution, no-host, swallowed
  resolution failure
- test_zernio_schema.py: ZernioConfigRequest bounds (api_key/timezone length,
  accounts count/platform/id) + PublishRequest.scheduled_for max_length
- test_status_logs_slice.py: GET /api/status caps the log tail at 500 lines

Host 437 passed/3 skipped. Note: batch-orphan-cleanup and reframe-save-500 paths
need async subprocess/queue mocking — left as a known test gap (logic manually
verified, exercised indirectly by integration).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01VDkEfcWP5suCdutKJKnvCM

Author: fralapo

tests/api/test_status_logs_slice.py

@@ -0,0 +1,50 @@
+"""GET /api/status caps the returned log tail.
+
+The per-job log buffer is bounded at the producer (job_worker.MAX_LOG_LINES),
+but the status response slices to the last 500 lines so a long job never ships
+a multi-thousand-line payload on every 2s poll. Driven with TestClient WITHOUT
+the lifespan (same pattern as test_job_controls).
+"""
+from fastapi.testclient import TestClient
+
+from clippyme.api import app as app_module
+
+JOB_ID = "22222222-2222-4222-8222-222222222222"
+ORIGIN = {"Origin": "http://localhost:5175"}
+
+
+def _client():
+    return TestClient(app_module.app, headers=ORIGIN)
+
+
+def test_status_logs_sliced_to_last_500():
+    app_module.jobs.pop(JOB_ID, None)
+    try:
+        app_module.jobs[JOB_ID] = {
+            "status": "processing",
+            "logs": [f"line {i}" for i in range(1500)],
+            "result": None,
+        }
+        r = _client().get(f"/api/status/{JOB_ID}")
+        assert r.status_code == 200
+        logs = r.json().get("logs", [])
+        assert len(logs) == 500
+        assert logs[-1] == "line 1499"
+        assert logs[0] == "line 1000"
+    finally:
+        app_module.jobs.pop(JOB_ID, None)
+
+
+def test_status_short_logs_unchanged():
+    app_module.jobs.pop(JOB_ID, None)
+    try:
+        app_module.jobs[JOB_ID] = {
+            "status": "processing",
+            "logs": ["a", "b", "c"],
+            "result": None,
+        }
+        r = _client().get(f"/api/status/{JOB_ID}")
+        assert r.status_code == 200
+        assert r.json().get("logs") == ["a", "b", "c"]
+    finally:
+        app_module.jobs.pop(JOB_ID, None)

tests/api/test_zernio_schema.py

@@ -0,0 +1,72 @@
+"""Bounds tests for ZernioConfigRequest + PublishRequest.scheduled_for.
+
+The Zernio config endpoint and publish schedule accept user-supplied strings
+that land in data/config.json and the scheduler. These pin the size/shape
+guards: unbounded api_key, oversized/odd accounts maps, and a giant
+scheduled_for string are all rejected at the Pydantic boundary; normal
+payloads pass.
+"""
+import pytest
+from pydantic import ValidationError
+
+from clippyme.api.schemas import PublishRequest, ZernioConfigRequest
+
+
+def test_zernio_accepts_normal_config():
+    req = ZernioConfigRequest(
+        api_key="sk_live_abc123",
+        accounts={"tiktok": "acc1", "youtube": "acc2", "instagram": None},
+        timezone="Europe/Rome",
+    )
+    assert req.accounts["tiktok"] == "acc1"
+
+
+def test_zernio_accepts_empty():
+    req = ZernioConfigRequest()
+    assert req.api_key is None
+
+
+def test_zernio_rejects_oversized_api_key():
+    with pytest.raises(ValidationError):
+        ZernioConfigRequest(api_key="x" * 513)
+
+
+def test_zernio_rejects_oversized_timezone():
+    with pytest.raises(ValidationError):
+        ZernioConfigRequest(timezone="z" * 65)
+
+
+def test_zernio_rejects_too_many_accounts():
+    with pytest.raises(ValidationError):
+        ZernioConfigRequest(accounts={f"k{i}": "v" for i in range(17)})
+
+
+def test_zernio_rejects_unknown_platform():
+    with pytest.raises(ValidationError):
+        ZernioConfigRequest(accounts={"myspace": "a"})
+
+
+def test_zernio_rejects_oversized_account_id():
+    with pytest.raises(ValidationError):
+        ZernioConfigRequest(accounts={"tiktok": "a" * 257})
+
+
+def test_zernio_rejects_nonstring_account_id():
+    with pytest.raises(ValidationError):
+        ZernioConfigRequest(accounts={"tiktok": {"nested": 1}})
+
+
+def test_publish_accepts_normal_scheduled_for():
+    req = PublishRequest(
+        platforms=[{"platform": "tiktok", "accountId": "a"}],
+        scheduled_for="2026-04-08T12:00:00",
+    )
+    assert req.scheduled_for.startswith("2026")
+
+
+def test_publish_rejects_oversized_scheduled_for():
+    with pytest.raises(ValidationError):
+        PublishRequest(
+            platforms=[{"platform": "tiktok", "accountId": "a"}],
+            scheduled_for="9" * 65,
+        )

tests/pipeline/test_download.py

@@ -1,9 +1,57 @@
 """Tests for clippyme.pipeline.download helpers (host-runnable; no network)."""
 import os
 
+import pytest
+
 from clippyme.pipeline import download as dl
 
 
+def _fake_getaddrinfo(*ips):
+    """Build a getaddrinfo stub returning the given IP strings."""
+    def _stub(host, port, *a, **k):
+        return [(None, None, None, "", (ip, 0)) for ip in ips]
+    return _stub
+
+
+def test_reject_rebound_literal_internal_ip_raises():
+    with pytest.raises(ValueError):
+        dl._reject_rebound_internal("http://127.0.0.1/video")
+
+
+def test_reject_rebound_literal_public_ip_passes():
+    # 8.8.8.8 is public — must not raise.
+    dl._reject_rebound_internal("http://8.8.8.8/video")
+
+
+def test_reject_rebound_all_internal_resolution_raises(monkeypatch):
+    monkeypatch.setattr(dl.socket, "getaddrinfo", _fake_getaddrinfo("192.168.1.10", "127.0.0.1"))
+    with pytest.raises(ValueError):
+        dl._reject_rebound_internal("http://rebind.evil.test/x")
+
+
+def test_reject_rebound_public_resolution_passes(monkeypatch):
+    monkeypatch.setattr(dl.socket, "getaddrinfo", _fake_getaddrinfo("93.184.216.34"))
+    dl._reject_rebound_internal("http://example.com/x")  # no raise
+
+
+def test_reject_rebound_mixed_public_and_internal_passes(monkeypatch):
+    # Not ALL internal → allowed (only blocks when every address is internal).
+    monkeypatch.setattr(dl.socket, "getaddrinfo", _fake_getaddrinfo("93.184.216.34", "10.0.0.1"))
+    dl._reject_rebound_internal("http://example.com/x")  # no raise
+
+
+def test_reject_rebound_no_host_returns_none():
+    assert dl._reject_rebound_internal("not a url") is None
+
+
+def test_reject_rebound_resolution_failure_is_swallowed(monkeypatch):
+    def _boom(*a, **k):
+        raise OSError("dns down")
+    monkeypatch.setattr(dl.socket, "getaddrinfo", _boom)
+    # Resolution hiccup must not block a legit download — yt-dlp handles it.
+    assert dl._reject_rebound_internal("http://example.com/x") is None
+
+
 def test_sanitize_filename_strips_invalid_chars():
     assert dl.sanitize_filename('a<b>c:d"e/f\\g|h?i*j') == "abcdefghij"