flan/hark
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

0.3.1: hardening pass — fix recanonicalize corruption, deploy crash-loop, POST desync

Browse Source 
A full audit before the TrueNAS deploy surfaced 15 issues, all fixed and
covered by regression tests:

- hark canon: label-only matches could clobber an unrelated topic's QID
  and merge distinct entities sharing a Wikidata display label; now only
  a QID match (or an unresolved same-label topic) merges, and genuine
  label collisions are disambiguated instead of crashing on the
  topics.label unique constraint.
- Web UI: a missing hark.db (fresh volume, no ingest yet) crashed every
  route with no HTTP response; now returns 503.
- Web UI: POST requests that redirected before reading the body left it
  undrained, desyncing the next HTTP/1.1 keep-alive request; body is now
  always consumed, and oversized bodies close the connection instead.
- Docker: non-root hark user couldn't write a freshly-created bind mount,
  crash-looping on first start; entrypoint now fixes ownership as root
  then drops to hark via gosu.
- hark load: per-record error isolation (one bad record no longer aborts
  the batch); re-loading already-extracted episodes reports as a skip,
  not a failure.
- Wikidata canonicalizer: Retry-After in HTTP-date form no longer crashes
  and gets silently swallowed as "no match"; transport errors (timeouts,
  connection resets) now retry like throttling responses do.
- Consolidated three near-duplicate topic-listing queries into one
  builder; deduped GENRES_FILTER against extract.GENRES.
This commit is contained in:
flan
2026-07-10 21:52:05 +00:00
parent 5e752f0277
commit ada07544f3
13 changed files with 453 additions and 188 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CHANGELOG.md
+42
View File
@@ -7,6 +7,48 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
## [Unreleased]
## [0.3.1] - 2026-07-10
Hardening pass ahead of the TrueNAS deploy: a full audit surfaced 15 issues,
all fixed and covered by regression tests.
### Fixed
- `hark canon`: a topic sharing another entity's Wikidata display label
(e.g. "Mercury" the planet vs. the element) could silently overwrite that
entity's QID and merge unrelated episodes onto it. Only an actual QID match
(or an unresolved same-label topic) is now treated as a merge target; a
genuine label collision between two resolved entities is disambiguated
with the QID instead of colliding on the `topics.label` unique constraint.
- Web UI: a missing or not-yet-created `hark.db` (e.g. a fresh volume before
the first ingest) crashed every authenticated route with an unhandled
exception and no HTTP response; now returns a clear 503.
- Web UI: POST requests that redirected before reading the body (expired
session, unmatched route) left it undrained, desyncing the next
HTTP/1.1 keep-alive request on the same connection; the body is now always
consumed. An oversized body now closes the connection instead of risking
the same desync.
- Docker: the non-root `hark` user couldn't write a freshly-created bind
mount or volume (Docker creates these as root), so the container
crash-looped on first start. The entrypoint now fixes ownership as root
before dropping to the unprivileged user via `gosu`.
- `hark load`: a malformed record no longer aborts the whole batch — each
record is isolated like `hark extract` already isolates episodes.
Re-loading already-extracted episodes is now reported as a skip, not a
failure (previously exit code 1 on an idempotent re-run).
- Wikidata canonicalizer: a `Retry-After` header in HTTP-date form (RFC 7231
permits either format) crashed and was silently swallowed as "no match";
parsing now handles both forms and caps the wait. Transport-level failures
(timeouts, connection resets) now retry like throttling responses do,
instead of giving up on the first attempt.
### Changed
- Consolidated three near-duplicate topic-listing queries (home, `/topics`,
`/search`, and `hark topics`) into one query builder; `/search`'s topic
results are now capped like every other list view.
- `GENRES_FILTER` in the web UI is no longer a second copy of `extract.GENRES`.
## [0.3.0] - 2026-07-10
Web frontend + deployment.
Dockerfile
+13 -2
View File
@@ -22,8 +22,19 @@ RUN --mount=type=cache,target=/root/.cache/uv uv sync --frozen --no-dev
ENV PATH="/app/.venv/bin:$PATH" \
HARK_DB=/app/data/hark.db \
HARK_AUTH_DB=/app/data/auth.db
# gosu drops from root to the unprivileged `hark` user after the entrypoint
# fixes ownership of /app/data — Docker creates bind mounts and anonymous
# volumes as root, which uid 8710 can't write to on its own.
RUN apt-get update && apt-get install -y --no-install-recommends gosu \
&& rm -rf /var/lib/apt/lists/* \
&& useradd --system --uid 8710 --no-create-home hark
COPY docker-entrypoint.sh /usr/local/bin/
RUN chmod +x /usr/local/bin/docker-entrypoint.sh
VOLUME ["/app/data"]
EXPOSE 8710
ENTRYPOINT ["hark"]
CMD ["web", "--bind", "0.0.0.0:8710"]
ENTRYPOINT ["docker-entrypoint.sh"]
CMD ["hark", "web", "--bind", "0.0.0.0:8710"]
docker-entrypoint.sh
Executable
+9
View File
@@ -0,0 +1,9 @@
#!/bin/sh
# Runs as root just long enough to fix ownership of the mounted /app/data
# (Docker creates bind mounts and anonymous volumes as root, which the
# unprivileged `hark` user can't write to), then execs into that user for
# everything else. No application code ever runs as root.
set -e
mkdir -p /app/data
chown -R hark:hark /app/data
exec gosu hark "$@"
pyproject.toml
+1 -1
View File
@@ -1,6 +1,6 @@
[project]
name = "hark"
version = "0.3.0"
version = "0.3.1"
description = "Cross-podcast topic index and discovery service"
readme = "README.md"
requires-python = ">=3.12"
src/hark/__init__.py
+1 -1
View File
@@ -1,3 +1,3 @@
"""hark — cross-podcast topic index and discovery service."""
__version__ = "0.3.0"
__version__ = "0.3.1"
src/hark/cli.py
+32 -32
View File
@@ -5,6 +5,7 @@ from __future__ import annotations
import argparse
import os
import sys
from typing import Callable
import httpx
@@ -22,6 +23,25 @@ def make_client() -> httpx.Client:
)
def make_reporter() -> tuple[Callable[[pipeline.ExtractResult], None], dict[str, int]]:
"""Shared per-episode progress printer for `extract` and `load`."""
counts = {"ok": 0, "failed": 0, "skipped": 0}
def report(r: pipeline.ExtractResult) -> None:
if r.error:
counts["failed"] += 1
print(f" FAIL {r.show} — {r.title}: {r.error}")
elif r.skipped:
counts["skipped"] += 1
print(f" skip {r.show} — {r.title}: already extracted")
else:
counts["ok"] += 1
labels = "; ".join(r.labels) if r.labels else "(no subject)"
print(f" ok {r.show} — {r.title} -> {labels}")
return report, counts
def cmd_resolve(args: argparse.Namespace) -> int:
conn = db.connect(args.db)
names = resolve.read_feeds_file(args.feeds)
@@ -80,17 +100,7 @@ def cmd_extract(args: argparse.Namespace) -> int:
return 1
extractor = extract.ClaudeExtractor(client, model=args.model)
ok = failed = 0
def report(r: pipeline.ExtractResult) -> None:
nonlocal ok, failed
if r.error:
failed += 1
print(f" FAIL {r.show} — {r.title}: {r.error}")
else:
ok += 1
labels = "; ".join(r.labels) if r.labels else "(no subject)"
print(f" ok {r.show} — {r.title} -> {labels}")
report, counts = make_reporter()
with make_client() as http_client:
canon = wikidata.Canonicalizer(http_client)
@@ -101,8 +111,8 @@ def cmd_extract(args: argparse.Namespace) -> int:
remaining = conn.execute(
"SELECT COUNT(*) FROM episodes WHERE extracted_at IS NULL"
).fetchone()[0]
print(f"extracted {ok} episodes ({failed} failed, {remaining} still pending)")
return 1 if failed else 0
print(f"extracted {counts['ok']} episodes ({counts['failed']} failed, {remaining} still pending)")
return 1 if counts["failed"] else 0
def cmd_load(args: argparse.Namespace) -> int:
@@ -115,13 +125,16 @@ def cmd_load(args: argparse.Namespace) -> int:
except (OSError, json.JSONDecodeError) as exc:
print(f"cannot read {args.file}: {exc}", file=sys.stderr)
return 1
ok = failed = 0
ok = failed = skipped = 0
def report(r: pipeline.ExtractResult) -> None:
nonlocal ok, failed
nonlocal ok, failed, skipped
if r.error:
failed += 1
print(f" FAIL {r.show} — {r.title}: {r.error}")
elif r.skipped:
skipped += 1
print(f" skip {r.show} — {r.title}: already extracted")
else:
ok += 1
labels = "; ".join(r.labels) if r.labels else "(no subject)"
@@ -132,7 +145,7 @@ def cmd_load(args: argparse.Namespace) -> int:
pipeline.load_extractions(
conn, records, canon.canonicalize, source=args.source, on_result=report
)
print(f"loaded {ok} episodes ({failed} failed)")
print(f"loaded {ok} episodes ({skipped} already loaded, {failed} failed)")
return 1 if failed else 0
@@ -152,23 +165,10 @@ def cmd_canon(args: argparse.Namespace) -> int:
def cmd_topics(args: argparse.Namespace) -> int:
from . import web # deferred: other commands work without importing the web module
conn = db.connect(args.db)
rows = conn.execute(
"""
SELECT t.label, t.wikidata_id,
COUNT(DISTINCT et.episode_id) AS episodes,
COUNT(DISTINCT e.show_id) AS shows,
COALESCE(GROUP_CONCAT(DISTINCT tg.genre), '') AS genres
FROM topics t
JOIN episode_topics et ON et.topic_id = t.id
JOIN episodes e ON e.id = et.episode_id
LEFT JOIN topic_genres tg ON tg.topic_id = t.id
GROUP BY t.id
ORDER BY shows DESC, episodes DESC, t.label
LIMIT ?
""",
(args.limit,),
).fetchall()
rows = conn.execute(*web.topics_query(limit=args.limit)).fetchall()
if not rows:
print("no topics yet — run `hark extract` first", file=sys.stderr)
return 1
src/hark/pipeline.py
+52 -22
View File
@@ -32,6 +32,7 @@ class ExtractResult:
title: str
labels: list[str] = field(default_factory=list)
error: str | None = None
skipped: bool = False # e.g. already extracted — not a failure, just a no-op
def pending_episodes(conn: sqlite3.Connection, limit: int | None = None) -> list[sqlite3.Row]:
@@ -153,24 +154,48 @@ def recanonicalize(
"""Retry Wikidata canonicalization for topics without a QID.
Used after throttled or offline runs. A fresh match either upgrades the
topic in place (label + QID) or, when another topic already owns that QID
or label, merges this topic into it: episode links and genres move over,
the duplicate row is deleted.
topic in place (label + QID) or, when another topic already owns that
QID, merges this topic into it: episode links and genres move over, the
duplicate row is deleted. A label collision with a topic that already
carries a *different* QID is not a merge — that's two distinct entities
sharing a display string (e.g. "Mercury" the planet vs. the element) —
so it's left as an in-place update rather than corrupting the existing
topic's identity.
"""
results: list[CanonResult] = []
rows = conn.execute("SELECT id, label FROM topics WHERE wikidata_id IS NULL").fetchall()
for row in rows:
for row in conn.execute("SELECT id, label FROM topics WHERE wikidata_id IS NULL").fetchall():
# Re-check: an earlier iteration in this same pass may have already
# merged this row away or resolved it as a merge target.
current = conn.execute(
"SELECT wikidata_id FROM topics WHERE id = ?", (row["id"],)
).fetchone()
if current is None or current["wikidata_id"] is not None:
continue
match = canonicalize(row["label"])
if match is None:
continue
target = conn.execute(
"SELECT id FROM topics WHERE (wikidata_id = ? OR label = ?) AND id != ?",
(match.qid, match.label, row["id"]),
"SELECT id FROM topics WHERE wikidata_id = ? AND id != ?",
(match.qid, row["id"]),
).fetchone()
if target is None:
target = conn.execute(
"SELECT id FROM topics WHERE label = ? AND wikidata_id IS NULL AND id != ?",
(match.label, row["id"]),
).fetchone()
if target is None:
# A different, already-resolved topic may already occupy this
# exact label (the "Mercury" planet/element case) — topics.label
# is UNIQUE, so adopting it verbatim would crash. Disambiguate
# with the QID instead of colliding or clobbering that topic.
label = match.label
if conn.execute(
"SELECT 1 FROM topics WHERE label = ? AND id != ?", (label, row["id"])
).fetchone():
label = f"{match.label} ({match.qid})"
conn.execute(
"UPDATE topics SET label = ?, wikidata_id = ? WHERE id = ?",
(match.label, match.qid, row["id"]),
(label, match.qid, row["id"]),
)
merged = False
else:
@@ -194,8 +219,9 @@ def recanonicalize(
)
conn.execute("DELETE FROM topics WHERE id = ?", (row["id"],))
merged = True
label = match.label
conn.commit()
results.append(CanonResult(row["label"], match.label, match.qid, merged))
results.append(CanonResult(row["label"], label, match.qid, merged))
return results
@@ -227,22 +253,26 @@ def load_extractions(
error=f"unknown episode_id {episode_id!r}")
elif row["extracted_at"] is not None:
result = ExtractResult(episode_id=row["id"], show=row["show"],
title=row["title"] or "", error="already extracted")
title=row["title"] or "", skipped=True)
else:
topics = [
ExtractedTopic(
label=str(t.get("label", "")).strip(),
genres=tuple(g for g in t.get("genres", ()) if g in GENRES),
confidence=None if t.get("confidence") is None
else max(0.0, min(1.0, float(t["confidence"]))),
)
for t in record.get("topics", [])
if str(t.get("label", "")).strip()
]
result = ExtractResult(episode_id=row["id"], show=row["show"],
title=row["title"] or "")
result.labels = _store(conn, row["id"], topics, canonicalize, source)
conn.commit()
try:
topics = [
ExtractedTopic(
label=str(t.get("label", "")).strip(),
genres=tuple(g for g in t.get("genres", ()) if g in GENRES),
confidence=None if t.get("confidence") is None
else max(0.0, min(1.0, float(t["confidence"]))),
)
for t in record.get("topics", [])
if str(t.get("label", "")).strip()
]
result.labels = _store(conn, row["id"], topics, canonicalize, source)
conn.commit()
except Exception as exc: # noqa: BLE001 — per-record isolation, keep the batch going
conn.rollback()
result.error = str(exc)
results.append(result)
if on_result:
on_result(result)
src/hark/web.py
+126 -113
View File
@@ -18,6 +18,7 @@ embedded stylesheet served from /static/style.css so the CSP can stay strict
from __future__ import annotations
import contextlib
import hashlib
import html
import os
@@ -31,15 +32,12 @@ from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from . import __version__
from .extract import GENRES as GENRES_FILTER
PW_ITERS = 120_000
SESSION_DAYS = 30
COOKIE = "hark_session"
GENRES_FILTER = (
"true_crime", "history", "disaster", "scam_fraud", "biography",
"espionage", "cult", "mystery", "other",
)
MAX_FORM_BYTES = 65536
AUTH_SCHEMA = """
CREATE TABLE IF NOT EXISTS users (
@@ -82,11 +80,10 @@ class Auth:
def __init__(self, path: str | Path, admin_token: str | None, admin_user: str = "admin"):
self.path = str(path)
self.admin_token = admin_token or None
conn = self._connect()
conn.executescript(AUTH_SCHEMA)
conn.execute("INSERT OR IGNORE INTO users (username) VALUES (?)", (admin_user,))
conn.commit()
conn.close()
with contextlib.closing(self._connect()) as conn:
conn.executescript(AUTH_SCHEMA)
conn.execute("INSERT OR IGNORE INTO users (username) VALUES (?)", (admin_user,))
conn.commit()
def _connect(self) -> sqlite3.Connection:
conn = sqlite3.connect(self.path)
@@ -97,8 +94,7 @@ class Auth:
"""Return user id on success. Fail-closed: an account with no stored
password only accepts the bootstrap admin token, and if that is unset
nothing is accepted."""
conn = self._connect()
try:
with contextlib.closing(self._connect()) as conn:
row = conn.execute(
"SELECT id, salt, password_hash FROM users WHERE username = ?", (username,)
).fetchone()
@@ -112,25 +108,21 @@ class Auth:
if self.admin_token and constant_eq(password, self.admin_token):
return row["id"]
return None
finally:
conn.close()
def create_session(self, user_id: int) -> str:
token = secrets.token_hex(32)
conn = self._connect()
conn.execute(
"INSERT INTO sessions (token, user_id, expires_at) VALUES (?, ?, ?)",
(token, user_id, iso(utcnow() + timedelta(days=SESSION_DAYS))),
)
conn.commit()
conn.close()
with contextlib.closing(self._connect()) as conn:
conn.execute(
"INSERT INTO sessions (token, user_id, expires_at) VALUES (?, ?, ?)",
(token, user_id, iso(utcnow() + timedelta(days=SESSION_DAYS))),
)
conn.commit()
return token
def session_user(self, token: str | None) -> sqlite3.Row | None:
if not token:
return None
conn = self._connect()
try:
with contextlib.closing(self._connect()) as conn:
return conn.execute(
"""
SELECT u.id, u.username FROM sessions s JOIN users u ON u.id = s.user_id
@@ -138,28 +130,24 @@ class Auth:
""",
(token, iso(utcnow())),
).fetchone()
finally:
conn.close()
def drop_session(self, token: str | None) -> None:
if not token:
return
conn = self._connect()
conn.execute("DELETE FROM sessions WHERE token = ?", (token,))
conn.commit()
conn.close()
with contextlib.closing(self._connect()) as conn:
conn.execute("DELETE FROM sessions WHERE token = ?", (token,))
conn.commit()
def set_password(self, user_id: int, password: str) -> None:
"""Set a new password and revoke every session (all devices log out)."""
salt = secrets.token_hex(16)
conn = self._connect()
conn.execute(
"UPDATE users SET salt = ?, password_hash = ? WHERE id = ?",
(salt, stretch(salt, password), user_id),
)
conn.execute("DELETE FROM sessions")
conn.commit()
conn.close()
with contextlib.closing(self._connect()) as conn:
conn.execute(
"UPDATE users SET salt = ?, password_hash = ? WHERE id = ?",
(salt, stretch(salt, password), user_id),
)
conn.execute("DELETE FROM sessions")
conn.commit()
# ---------------------------------------------------------------------------
@@ -263,7 +251,7 @@ class App:
GROUP BY et.topic_id HAVING COUNT(DISTINCT e.show_id) > 1)
"""
).fetchone()[0]
rows = conn.execute(TOP_TOPICS_SQL + " LIMIT 15").fetchall()
rows = conn.execute(*topics_query(limit=15)).fetchall()
finally:
conn.close()
cards = f"""
@@ -284,13 +272,11 @@ class App:
def view_topics(self, user, params) -> str:
genre = params.get("genre", [""])[0]
if genre not in GENRES_FILTER:
genre = ""
conn = self.db()
try:
if genre in GENRES_FILTER:
rows = conn.execute(TOP_TOPICS_BY_GENRE_SQL, (genre,)).fetchall()
else:
genre = ""
rows = conn.execute(TOP_TOPICS_SQL + " LIMIT 200").fetchall()
rows = conn.execute(*topics_query(genre=genre, limit=200)).fetchall()
finally:
conn.close()
pills = " ".join(
@@ -350,10 +336,7 @@ class App:
like = f"%{q}%"
conn = self.db()
try:
topics = conn.execute(
TOP_TOPICS_SQL.replace("GROUP BY", "WHERE t.label LIKE ? COLLATE NOCASE OR t.wikidata_id = ? GROUP BY"),
(like, q),
).fetchall()
topics = conn.execute(*topics_query(q=q, limit=200)).fetchall()
episodes = conn.execute(
"""
SELECT e.id, e.title, e.pubdate, COALESCE(s.title, s.query) AS show
@@ -418,33 +401,31 @@ class App:
return page("account", body, user["username"])
TOP_TOPICS_SQL = """
SELECT t.id, t.label, t.wikidata_id,
COUNT(DISTINCT et.episode_id) AS episodes,
COUNT(DISTINCT e.show_id) AS shows,
COALESCE(GROUP_CONCAT(DISTINCT tg.genre), '') AS genres
FROM topics t
JOIN episode_topics et ON et.topic_id = t.id
JOIN episodes e ON e.id = et.episode_id
LEFT JOIN topic_genres tg ON tg.topic_id = t.id
GROUP BY t.id
ORDER BY shows DESC, episodes DESC, t.label
"""
TOP_TOPICS_BY_GENRE_SQL = """
SELECT t.id, t.label, t.wikidata_id,
COUNT(DISTINCT et.episode_id) AS episodes,
COUNT(DISTINCT e.show_id) AS shows,
COALESCE(GROUP_CONCAT(DISTINCT tg.genre), '') AS genres
FROM topics t
JOIN episode_topics et ON et.topic_id = t.id
JOIN episodes e ON e.id = et.episode_id
LEFT JOIN topic_genres tg ON tg.topic_id = t.id
WHERE t.id IN (SELECT topic_id FROM topic_genres WHERE genre = ?)
GROUP BY t.id
ORDER BY shows DESC, episodes DESC, t.label
LIMIT 200
"""
def topics_query(genre: str = "", q: str = "", limit: int | None = None) -> tuple[str, tuple]:
"""Build the shared topic-listing query: base coverage stats, optionally
filtered by genre or a label/QID search term, optionally capped."""
sql = """
SELECT t.id, t.label, t.wikidata_id,
COUNT(DISTINCT et.episode_id) AS episodes,
COUNT(DISTINCT e.show_id) AS shows,
COALESCE(GROUP_CONCAT(DISTINCT tg.genre), '') AS genres
FROM topics t
JOIN episode_topics et ON et.topic_id = t.id
JOIN episodes e ON e.id = et.episode_id
LEFT JOIN topic_genres tg ON tg.topic_id = t.id
"""
params: list = []
if genre:
sql += " WHERE t.id IN (SELECT topic_id FROM topic_genres WHERE genre = ?)"
params.append(genre)
elif q:
sql += " WHERE (t.label LIKE ? COLLATE NOCASE OR t.wikidata_id = ?)"
params.extend([f"%{q}%", q])
sql += " GROUP BY t.id ORDER BY shows DESC, episodes DESC, t.label"
if limit:
sql += " LIMIT ?"
params.append(limit)
return sql, tuple(params)
def conf(value) -> str:
@@ -453,8 +434,11 @@ def conf(value) -> str:
def episode_cell(row) -> str:
title = esc(row["title"])
if row["audio_url"]:
return f'{title} <a class="qid" href="{esc(row["audio_url"])}" rel="noreferrer">▶</a>'
url = row["audio_url"] or ""
# Enclosure URLs come from third-party feeds; only link plain http(s) so a
# hostile feed can't smuggle a javascript:/data: scheme into an href.
if urllib.parse.urlsplit(url).scheme in ("http", "https"):
return f'{title} <a class="qid" href="{esc(url)}" rel="noreferrer">▶</a>'
return title
@@ -490,9 +474,12 @@ class Handler(BaseHTTPRequestHandler):
server_version = f"hark/{__version__}"
protocol_version = "HTTP/1.1"
def log_message(self, fmt, *args): # quiet access log; errors still surface
def log_message(self, fmt, *args): # quiet access log
pass
def log_error(self, fmt, *args): # bypass log_message so errors still surface
BaseHTTPRequestHandler.log_message(self, fmt, *args)
# -- helpers -------------------------------------------------------------
def _security_headers(self):
@@ -534,9 +521,26 @@ class Handler(BaseHTTPRequestHandler):
def form(self) -> dict:
length = int(self.headers.get("Content-Length", 0) or 0)
raw = self.rfile.read(min(length, 65536)).decode(errors="replace")
if length > MAX_FORM_BYTES:
# Too large to read safely; don't try to keep the connection
# alive with an unread tail still sitting in the socket.
self.close_connection = True
return {}
raw = self.rfile.read(length).decode(errors="replace")
return {k: v[0] for k, v in urllib.parse.parse_qs(raw).items()}
def not_found(self, user) -> None:
return self.respond(404, page("404", "<h1>Not found</h1>", user["username"]))
def db_unavailable(self, user) -> None:
return self.respond(503, page(
"unavailable",
"<h1>Not ready</h1><p>The topic database hasn't been created yet — "
"run <code>hark ingest</code> and <code>hark extract</code> "
"(or <code>hark load</code>) first.</p>",
user["username"],
))
# -- routing -------------------------------------------------------------
def do_GET(self):
@@ -556,33 +560,36 @@ class Handler(BaseHTTPRequestHandler):
if user is None:
return self.redirect("/login")
if route == "/":
return self.respond(200, app.view_home(user))
if route == "/topics":
return self.respond(200, app.view_topics(user, params))
if route.startswith("/topic/"):
try:
topic_id = int(route.rsplit("/", 1)[1])
except ValueError:
return self.respond(404, page("404", "<h1>Not found</h1>", user["username"]))
body = app.view_topic(user, topic_id)
if body is None:
return self.respond(404, page("404", "<h1>Not found</h1>", user["username"]))
return self.respond(200, body)
if route == "/search":
return self.respond(200, app.view_search(user, params))
if route == "/shows":
return self.respond(200, app.view_shows(user))
if route == "/account":
return self.respond(200, app.view_account(user))
return self.respond(404, page("404", "<h1>Not found</h1>", user["username"]))
try:
if route == "/":
return self.respond(200, app.view_home(user))
if route == "/topics":
return self.respond(200, app.view_topics(user, params))
if route.startswith("/topic/"):
try:
topic_id = int(route.rsplit("/", 1)[1])
except ValueError:
return self.not_found(user)
body = app.view_topic(user, topic_id)
if body is None:
return self.not_found(user)
return self.respond(200, body)
if route == "/search":
return self.respond(200, app.view_search(user, params))
if route == "/shows":
return self.respond(200, app.view_shows(user))
if route == "/account":
return self.respond(200, app.view_account(user))
except sqlite3.OperationalError:
return self.db_unavailable(user)
return self.not_found(user)
def do_POST(self):
app = self.app
route = self.path.rstrip("/")
form = self.form() # always drain the body, even on routes that ignore it
if route == "/login":
form = self.form()
user_id = app.auth.verify(form.get("username", ""), form.get("password", ""))
if user_id is None:
time.sleep(0.4) # blunt brute-force throttle
@@ -595,19 +602,21 @@ class Handler(BaseHTTPRequestHandler):
if user is None:
return self.redirect("/login")
if route == "/logout":
app.auth.drop_session(self.cookie_token())
return self.redirect("/login", {"Set-Cookie": app.cookie_attrs("", 0)})
if route == "/account/password":
form = self.form()
pw, pw2 = form.get("password", ""), form.get("password2", "")
if len(pw) < 8:
return self.respond(400, app.view_account(user, err="Password too short (min 8)."))
if pw != pw2:
return self.respond(400, app.view_account(user, err="Passwords do not match."))
app.auth.set_password(user["id"], pw)
return self.redirect("/login", {"Set-Cookie": app.cookie_attrs("", 0)})
return self.respond(404, page("404", "<h1>Not found</h1>", user["username"]))
try:
if route == "/logout":
app.auth.drop_session(self.cookie_token())
return self.redirect("/login", {"Set-Cookie": app.cookie_attrs("", 0)})
if route == "/account/password":
pw, pw2 = form.get("password", ""), form.get("password2", "")
if len(pw) < 8:
return self.respond(400, app.view_account(user, err="Password too short (min 8)."))
if pw != pw2:
return self.respond(400, app.view_account(user, err="Passwords do not match."))
app.auth.set_password(user["id"], pw)
return self.redirect("/login", {"Set-Cookie": app.cookie_attrs("", 0)})
except sqlite3.OperationalError:
return self.db_unavailable(user)
return self.not_found(user)
def make_server(db_path: str | Path, auth_path: str | Path, bind: str = "0.0.0.0:8710",
@@ -615,8 +624,12 @@ def make_server(db_path: str | Path, auth_path: str | Path, bind: str = "0.0.0.0
auth = Auth(auth_path, admin_token=admin_token)
app = App(db_path, auth, cookie_secure=cookie_secure)
host, _, port = bind.rpartition(":")
try:
port_num = int(port)
except ValueError:
raise SystemExit(f"invalid --bind {bind!r}: expected host:port or :port")
handler = type("BoundHandler", (Handler,), {"app": app})
return ThreadingHTTPServer((host or "0.0.0.0", int(port)), handler)
return ThreadingHTTPServer((host or "0.0.0.0", port_num), handler)
def serve(db_path: str | Path, auth_path: str | Path, bind: str, admin_token: str | None,
src/hark/wikidata.py
+43 -15
View File
@@ -8,12 +8,32 @@ episodes, recurring subjects) to one request each.
from __future__ import annotations
import email.utils
import time
from dataclasses import dataclass
from datetime import datetime, timezone
import httpx
API_URL = "https://www.wikidata.org/w/api.php"
MAX_BACKOFF = 30.0
def _retry_after_seconds(value: str | None, default: float = 5.0) -> float:
"""Parse a Retry-After header: either delta-seconds or an HTTP-date
(both valid per RFC 7231), capped so a hostile/misconfigured server
can't stall a bulk run for arbitrarily long."""
if not value:
return default
try:
seconds = float(value)
except ValueError:
try:
when = email.utils.parsedate_to_datetime(value)
except (TypeError, ValueError):
return default
seconds = (when - datetime.now(timezone.utc)).total_seconds()
return max(0.0, min(seconds, MAX_BACKOFF))
@dataclass
@@ -30,7 +50,7 @@ class Canonicalizer:
def __init__(self, client: httpx.Client, delay: float = 0.25, retries: int = 2):
self.client = client
self.delay = delay
self.retries = retries
self.retries = max(0, retries)
self._cache: dict[str, WikidataMatch | None] = {}
def canonicalize(self, label: str) -> WikidataMatch | None:
@@ -42,23 +62,31 @@ class Canonicalizer:
return self._cache[key]
def _get(self, label: str) -> httpx.Response:
last_exc: httpx.TransportError | None = None
for attempt in range(self.retries + 1):
resp = self.client.get(
API_URL,
params={
"action": "wbsearchentities",
"search": label,
"language": "en",
"type": "item",
"limit": 1,
"format": "json",
},
)
if resp.status_code in (429, 500, 502, 503) and attempt < self.retries:
time.sleep(float(resp.headers.get("retry-after", 5)))
try:
resp = self.client.get(
API_URL,
params={
"action": "wbsearchentities",
"search": label,
"language": "en",
"type": "item",
"limit": 1,
"format": "json",
},
)
except httpx.TransportError as exc:
last_exc = exc
if attempt < self.retries:
time.sleep(self.delay or 1.0)
continue
raise
if resp.status_code in (429, 500, 502, 503, 504) and attempt < self.retries:
time.sleep(_retry_after_seconds(resp.headers.get("retry-after")))
continue
return resp
return resp
raise last_exc if last_exc else AssertionError("unreachable: retries >= 0 always iterates")
def _lookup(self, label: str) -> WikidataMatch | None:
try:
tests/test_pipeline.py
+26 -1
View File
@@ -138,7 +138,8 @@ def test_load_extractions_validates_and_stores(tmp_path):
assert pipeline.pending_episodes(conn) == [] # both marked, incl. zero-topic
again = pipeline.load_extractions(conn, records[:1], canonicalize, source="sess")
assert again[0].error == "already extracted"
assert again[0].error is None
assert again[0].skipped is True
def test_recanonicalize_upgrades_in_place(tmp_path):
@@ -181,6 +182,30 @@ def test_recanonicalize_merges_duplicates(tmp_path):
assert genres == {"true_crime", "biography"}
def test_recanonicalize_does_not_clobber_unrelated_qid_on_label_collision(tmp_path):
"""Two distinct entities sharing a display label (planet vs. element,
both called "Mercury") must not get merged just because their labels
coincide — only a QID match (or an unresolved label match) may merge."""
conn = db.connect(tmp_path / "t.db")
seed(conn, ["ep1", "ep2"])
extractor = FakeExtractor({
"ep1": [ExtractedTopic(label="the planet", genres=("history",))],
"ep2": [ExtractedTopic(label="quicksilver metal", genres=("history",))],
})
pipeline.extract_pending(conn, extractor, lambda label: None, source="test")
matches = {
"the planet": WikidataMatch(qid="Q308", label="Mercury"),
"quicksilver metal": WikidataMatch(qid="Q925", label="Mercury"),
}
pipeline.recanonicalize(conn, lambda label: matches[label.casefold()])
rows = {r["wikidata_id"]: r["label"] for r in conn.execute("SELECT label, wikidata_id FROM topics")}
assert set(rows) == {"Q308", "Q925"} # two rows, neither QID overwritten
assert list(rows.values()).count("Mercury") == 1 # one keeps the plain label
assert any(v.startswith("Mercury (Q9") for v in rows.values()) # the other is disambiguated
def test_recanonicalize_skips_unmatched(tmp_path):
conn = db.connect(tmp_path / "t.db")
seed(conn, ["ep1"])
tests/test_web.py
+68
View File
@@ -124,6 +124,42 @@ def test_fail_closed_without_token(tmp_path):
srv.shutdown()
def test_missing_hark_db_returns_503_not_crash(tmp_path):
# hark.db is never created — simulates a fresh volume before any ingest.
srv = web.make_server(tmp_path / "hark.db", tmp_path / "auth.db",
bind="127.0.0.1:0", admin_token="t")
thread = threading.Thread(target=srv.serve_forever, daemon=True)
thread.start()
try:
cookie = login(srv, password="t")
resp, body = request(srv, "GET", "/", cookie=cookie)
assert resp.status == 503
assert "Not ready" in body
# /healthz never touches hark.db, so it stays healthy regardless
resp, _ = request(srv, "GET", "/healthz")
assert resp.status == 200
finally:
srv.shutdown()
def test_post_without_session_drains_body_no_keepalive_desync(server):
conn = http.client.HTTPConnection("127.0.0.1", server.server_address[1], timeout=5)
body = urllib.parse.urlencode({"password": "x", "password2": "x"})
conn.request("POST", "/account/password", body=body,
headers={"Content-Type": "application/x-www-form-urlencoded"})
resp1 = conn.getresponse()
assert resp1.status == 303 # no session -> redirected to /login
resp1.read()
# If the body above wasn't drained, this next request on the same
# connection would be parsed starting mid-body and come back mangled.
conn.request("GET", "/healthz")
resp2 = conn.getresponse()
assert resp2.status == 200
assert resp2.read() == b"ok"
conn.close()
def test_password_change_revokes_sessions_and_disables_token(server):
cookie = login(server)
resp, _ = request(server, "POST", "/account/password",
@@ -172,6 +208,38 @@ def test_cookie_secure_flag(tmp_path):
srv.shutdown()
def test_audio_link_scheme_allowlist(tmp_path):
conn = db.connect(tmp_path / "hark.db")
conn.execute("INSERT INTO shows (query, title, feed_url) VALUES ('q', 'S', 'http://x')")
conn.executemany(
"INSERT INTO episodes (show_id, guid, title, audio_url, extracted_at)"
" VALUES (1, ?, ?, ?, '2026-01-01T00:00:00Z')",
[
("g1", "ok", "https://cdn/a.mp3"),
("g2", "evil", "javascript:alert(1)"),
],
)
conn.execute("INSERT INTO topics (label) VALUES ('T')")
conn.executemany(
"INSERT INTO episode_topics (episode_id, topic_id, source) VALUES (?, 1, 't')",
[(1,), (2,)],
)
conn.commit()
conn.close()
srv = web.make_server(tmp_path / "hark.db", tmp_path / "auth.db",
bind="127.0.0.1:0", admin_token="t")
thread = threading.Thread(target=srv.serve_forever, daemon=True)
thread.start()
try:
resp, _ = request(srv, "POST", "/login", body={"username": "admin", "password": "t"})
cookie = resp.getheader("Set-Cookie").split(";")[0]
resp, body = request(srv, "GET", "/topic/1", cookie=cookie)
assert 'href="https://cdn/a.mp3"' in body
assert "javascript:" not in body
finally:
srv.shutdown()
def test_html_escapes_labels(tmp_path):
conn = db.connect(tmp_path / "hark.db")
conn.execute("INSERT INTO shows (query, title, feed_url) VALUES ('q', '<b>S</b>', 'http://x')")
tests/test_wikidata.py
+39
View File
@@ -75,3 +75,42 @@ def test_canonicalize_gives_up_after_retries():
canon = make_canon(handler, retries=2)
assert canon.canonicalize("Anything") is None
assert len(calls) == 3
def test_canonicalize_retries_http_date_retry_after(monkeypatch):
from hark import wikidata
calls = []
sleeps = []
monkeypatch.setattr(wikidata.time, "sleep", lambda s: sleeps.append(s))
def handler(request):
calls.append(1)
if len(calls) == 1:
# RFC 7231 permits an HTTP-date here, not just delta-seconds.
return httpx.Response(429, headers={"retry-after": "Wed, 21 Oct 2099 07:28:00 GMT"})
return search_response([{"id": "Q1", "label": "Whatever"}])
canon = make_canon(handler)
match = canon.canonicalize("whatever")
assert match.qid == "Q1"
assert len(calls) == 2
assert sleeps == [wikidata.MAX_BACKOFF] # far-future date clamps to the cap, doesn't crash
def test_canonicalize_retries_transport_errors(monkeypatch):
from hark import wikidata
monkeypatch.setattr(wikidata.time, "sleep", lambda s: None)
calls = []
def handler(request):
calls.append(1)
if len(calls) == 1:
raise httpx.ConnectError("boom", request=request)
return search_response([{"id": "Q99", "label": "Recovered"}])
canon = make_canon(handler)
match = canon.canonicalize("flaky")
assert match.qid == "Q99"
assert len(calls) == 2
uv.lock
Generated
+1 -1
View File
@@ -102,7 +102,7 @@ wheels = [
[[package]]
name = "hark"
version = "0.3.0"
version = "0.3.1"
source = { editable = "." }
dependencies = [
{ name = "anthropic" },