diff options
Diffstat (limited to 'packages/meshbay-hub/src/meshbay_hub')
| -rw-r--r-- | packages/meshbay-hub/src/meshbay_hub/app.py | 2 | ||||
| -rw-r--r-- | packages/meshbay-hub/src/meshbay_hub/csam.py | 159 |
2 files changed, 161 insertions, 0 deletions
diff --git a/packages/meshbay-hub/src/meshbay_hub/app.py b/packages/meshbay-hub/src/meshbay_hub/app.py index dcb255b..1c494fa 100644 --- a/packages/meshbay-hub/src/meshbay_hub/app.py +++ b/packages/meshbay-hub/src/meshbay_hub/app.py @@ -27,6 +27,7 @@ from meshbay_hub.api.groups import router as groups_router from meshbay_hub.api.revocation import router as revocation_router from meshbay_hub.api.moderation import router as moderation_router from meshbay_hub.api.federation import router as federation_router +from meshbay_hub.csam import csam_router from meshbay_hub.api.relay import router as relay_router from meshbay_hub.api.webapp import router as webapp_router from meshbay_hub.api.middleware import limiter @@ -72,6 +73,7 @@ def create_app(cfg: HubConfig | None = None) -> FastAPI: app.include_router(revocation_router) app.include_router(moderation_router) app.include_router(federation_router) + app.include_router(csam_router) app.include_router(relay_router) app.include_router(webapp_router) diff --git a/packages/meshbay-hub/src/meshbay_hub/csam.py b/packages/meshbay-hub/src/meshbay_hub/csam.py new file mode 100644 index 0000000..4dd0aed --- /dev/null +++ b/packages/meshbay-hub/src/meshbay_hub/csam.py @@ -0,0 +1,159 @@ +""" +MeshBay Hub — CSAM hash matching. + +Checks public content hashes against known CSAM (Child Sexual Abuse Material) +hash databases before allowing content to be registered or served publicly. + +Production integration: + - NCMEC (National Center for Missing & Exploited Children): PhotoDNA hash database + Access requires formal application: https://www.missingkids.org/gethelpnow/cybertipline + - IWF (Internet Watch Foundation): URL and hash list (UK-based) + Access via IWF membership: https://www.iwf.org.uk/our-technology/our-products/hash-list/ + +This module provides: + 1. A local CSAM hash database (SQLite file, populated from official sources) + 2. A check function used before content registration + 3. An admin endpoint to update the hash list + +IMPORTANT: Never log matched hashes or file contents. CSAM detection +must be reported to NCMEC (US law) or relevant authority immediately. +""" + +import hashlib +import logging +import os +from pathlib import Path + +log = logging.getLogger(__name__) + +# Default path for the CSAM hash database (blake3 hex hashes, one per line) +DEFAULT_CSAM_DB_PATH = Path("/var/lib/meshbay/hub/csam_hashes.txt") + + +class CSAMChecker: + """ + Checks content hashes against a known CSAM hash database. + + Usage: + checker = CSAMChecker() + checker.load() + if checker.is_known_csam(blake3_hex): + # refuse to serve, report to authority + pass + """ + + def __init__(self, db_path: Path = DEFAULT_CSAM_DB_PATH): + self._db_path = db_path + self._hashes: set[str] = set() + self._loaded = False + + def load(self, db_path: Path | None = None) -> int: + """ + Load CSAM hashes from the hash database file. + Returns the number of hashes loaded. + + File format: one blake3 hex hash per line (64 chars), comments with #. + """ + path = db_path or self._db_path + if not path.exists(): + log.warning("CSAM hash database not found: %s. " + "Contact NCMEC (US) or IWF (EU) for access.", path) + self._loaded = True + return 0 + + count = 0 + with open(path) as f: + for line in f: + line = line.strip() + if line and not line.startswith("#") and len(line) == 64: + self._hashes.add(line.lower()) + count += 1 + + self._loaded = True + log.info("CSAM hash database loaded: %d hashes from %s", count, path) + return count + + def is_known_csam(self, content_hash_hex: str) -> bool: + """ + Return True if the hash matches a known CSAM hash. + NEVER logs the hash or any file information. + """ + if not self._loaded: + self.load() + return content_hash_hex.lower() in self._hashes + + @property + def hash_count(self) -> int: + return len(self._hashes) + + def add_hash(self, hash_hex: str) -> None: + """Add a hash to the in-memory set (and optionally persist).""" + self._hashes.add(hash_hex.lower()) + + def update_from_file(self, new_db_path: Path) -> int: + """Hot-reload from a new hash database file.""" + old_count = len(self._hashes) + self._hashes.clear() + count = self.load(new_db_path) + log.info("CSAM database updated: %d → %d hashes", old_count, count) + return count + + +# Module-level singleton (initialised in hub lifespan) +_checker = CSAMChecker() + + +def get_csam_checker() -> CSAMChecker: + return _checker + + +def check_content_hash(blake3_hex: str) -> bool: + """ + Check a content hash against the CSAM database. + Returns True if the content is KNOWN CSAM — block immediately. + + Callers MUST: + 1. Refuse to serve the content + 2. Log the event (without the hash) for legal audit purposes + 3. Report to NCMEC CyberTipline if operating in the US: + https://www.missingkids.org/gethelpnow/cybertipline + """ + return _checker.is_known_csam(blake3_hex) + + +# ── Hub API integration ─────────────────────────────────────────────────────── + +from fastapi import APIRouter, Depends, HTTPException, UploadFile, File +from meshbay_hub.api.deps import get_current_user +from meshbay_hub.db.models import User + +csam_router = APIRouter(prefix="/v1/admin/csam", tags=["csam"]) + + +@csam_router.get("/status") +async def csam_status(current_user: User = Depends(get_current_user)): + """Return CSAM checker status (hash count, database path).""" + return { + "hash_count": _checker.hash_count, + "db_path": str(_checker._db_path), + "loaded": _checker._loaded, + "note": "Contact NCMEC or IWF for hash database access.", + } + + +@csam_router.post("/check") +async def check_hash( + body: dict, + current_user: User = Depends(get_current_user), +): + """ + Check a single hash. Admin use only. + Returns True/False WITHOUT logging the hash (legal requirement). + """ + hash_hex = body.get("hash", "") + if len(hash_hex) != 64: + raise HTTPException(status_code=422, detail="hash must be 64 hex chars") + matched = check_content_hash(hash_hex) + # Do NOT log whether a match was found — only log the check attempt + log.info("CSAM check performed by admin %s", current_user.username) + return {"matched": matched} |