1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
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 require_admin
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(require_admin)):
"""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(require_admin),
):
"""
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}
|