fix(malware-guard): tighten .url bookmark and Ren'Py exemptions and remove Spanish readme false-positive
9008394 verified | """Multi-Layer CSAM & Harmful Content Protection Engine for Armiger DHT Sync. | |
| Enforces: | |
| 1. Multi-stage ContentFilter normalization parity with multi-stage text normalization, leetspeak deobfuscation, | |
| zero-width character stripping, and pattern matching. | |
| 2. Deterministic & structural pre-ingest CSAM/abuse blocker checking title, file trees, | |
| file extensions, and disguised malware droppers before any shard serialization. | |
| 3. Ed25519-signed blinded SHA-256 tombstone purge list with monotonic sequence enforcement | |
| and cryptographic revocation verification. | |
| 4. Mandatory Armiger Mondrian conformal safety gate evaluation (junk/malware, out-of-scope, | |
| and strict adult namespace isolation). | |
| """ | |
| from __future__ import annotations | |
| import dataclasses | |
| import enum | |
| import functools | |
| import hashlib | |
| import hmac | |
| import json | |
| import math | |
| import os | |
| import re | |
| import time | |
| import unicodedata | |
| from typing import Any, List, Optional, Sequence, Tuple | |
| try: | |
| from cryptography.exceptions import InvalidSignature | |
| from cryptography.hazmat.primitives.asymmetric.ed25519 import ( | |
| Ed25519PrivateKey, | |
| Ed25519PublicKey, | |
| ) | |
| except ImportError: # pragma: no cover - stdlib fallback for bare CI | |
| InvalidSignature = Exception # type: ignore[misc,assignment] | |
| Ed25519PrivateKey = None # type: ignore[misc,assignment] | |
| Ed25519PublicKey = None # type: ignore[misc,assignment] | |
| class CSAMViolationType(str, enum.Enum): | |
| """Specific categories of content filter violations.""" | |
| CSAM_EXPLOITATION = "csam_exploitation" | |
| UNDERAGE_SEXUAL_CONTENT = "underage_sexual_content" | |
| NONCONSENSUAL_ABUSE = "nonconsensual_abuse" | |
| EXTREME_VIOLENCE_GORE = "extreme_violence_gore" | |
| MALWARE_PAYLOAD = "malware_payload" | |
| TOMBSTONED_HASH = "tombstoned_hash" | |
| SAFETY_GATE_JUNK = "safety_gate_junk" | |
| SAFETY_GATE_OUT_OF_SCOPE = "safety_gate_out_of_scope" | |
| UNISOLATED_ADULT = "unisolated_adult" | |
| class CSAMViolationError(Exception): | |
| """Raised when torrent metadata or payload violates CSAM/abuse safety policies.""" | |
| def __init__(self, violation_type: CSAMViolationType, reason: str, pattern: str | None = None) -> None: | |
| super().__init__(f"Safety violation [{violation_type.value}]: {reason}") | |
| self.violation_type = violation_type | |
| self.reason = reason | |
| self.pattern = pattern | |
| class ContentFilterResult: | |
| """Detailed result of content filter inspection.""" | |
| is_safe: bool | |
| violation_type: CSAMViolationType | None | |
| reason: str | None | |
| matched_pattern: str | None | |
| risk_score: float | |
| sanitized_title: str | |
| # --------------------------------------------------------------------------- | |
| # Leetspeak & Normalization Dictionaries | |
| # --------------------------------------------------------------------------- | |
| _LEET_MAP: dict[str, str] = { | |
| "0": "o", | |
| "1": "i", | |
| "3": "e", | |
| "4": "a", | |
| "@": "a", | |
| "$": "s", | |
| "5": "s", | |
| "7": "t", | |
| "+": "t", | |
| "!": "i", | |
| "8": "b", | |
| "9": "g", | |
| "|": "l", | |
| } | |
| # Confusable Cyrillic / Greek homoglyphs mapped to ASCII Latin equivalents | |
| _HOMOGLYPH_MAP: dict[str, str] = { | |
| "\u0430": "a", # Cyrillic a | |
| "\u0432": "b", # Cyrillic ve | |
| "\u0435": "e", # Cyrillic ie | |
| "\u0438": "i", # Cyrillic i | |
| "\u0456": "i", # Cyrillic byelorussian-ukrainian i | |
| "\u0457": "i", # Cyrillic yi | |
| "\u0458": "j", # Cyrillic je | |
| "\u043a": "k", # Cyrillic ka | |
| "\u043c": "m", # Cyrillic em | |
| "\u043d": "h", # Cyrillic en | |
| "\u043e": "o", # Cyrillic o | |
| "\u0440": "p", # Cyrillic er | |
| "\u0441": "c", # Cyrillic es | |
| "\u0455": "s", # Cyrillic dze | |
| "\u0442": "t", # Cyrillic te | |
| "\u0443": "y", # Cyrillic u | |
| "\u0445": "x", # Cyrillic ha | |
| "\u03b1": "a", # Greek alpha | |
| "\u03b2": "b", # Greek beta | |
| "\u03b5": "e", # Greek epsilon | |
| "\u03b9": "i", # Greek iota | |
| "\u03ba": "k", # Greek kappa | |
| "\u03bd": "v", # Greek nu | |
| "\u03bf": "o", # Greek omicron | |
| "\u03c1": "p", # Greek rho | |
| "\u03c4": "t", # Greek tau | |
| "\u03c5": "u", # Greek upsilon | |
| "\u03c7": "x", # Greek chi | |
| "\U0001d455": "h", # Reserved slot between math italic g (1D454) and i (1D456) | |
| } | |
| # Regex to strip zero-width characters, soft hyphens, word joiners, and bidi overrides/isolates | |
| _ZERO_WIDTH_RE = re.compile( | |
| r"[\u00AD\u034F\u061C\u180E\u200B-\u200F\u202A-\u202E\u2060-\u206F\uFEFF]" | |
| ) | |
| # Collapse obfuscated single-character runs (e.g., "c h i l d" -> "child") | |
| _SINGLE_CHAR_RUN_RE = re.compile(r"\b([a-z0-9])\s+(?=[a-z0-9]\b)") | |
| # Delimiters and non-alphanumeric punctuation | |
| _PUNCT_RE = re.compile(r"[\._\-\+\[\]\(\)\{\}\\\/\|\:\,\;\*\^\~\`\'\"]+") | |
| # Repeated characters (e.g. "teeeen" -> "teen") | |
| _REPEATED_CHARS_RE = re.compile(r"(.)\1{2,}") | |
| # Private tracker passkey / auth tokens in torrent titles | |
| _TRACKER_TAGS_RE = re.compile( | |
| r"(\[passkey=[^\]]+\]|\?passkey=[a-zA-Z0-9]+|\[auth=[^\]]+\]|torrentpass=[a-zA-Z0-9]+)", | |
| re.IGNORECASE, | |
| ) | |
| _HOMOGLYPH_TRANS = str.maketrans(_HOMOGLYPH_MAP) | |
| _LEET_TRANS = str.maketrans(_LEET_MAP) | |
| _PRESERVE_ALNUM_TOKENS = frozenset({ | |
| "1080p", "720p", "2160p", "480p", "1080i", "720i", | |
| "x264", "x265", "h264", "h265", "av1", "vp8", "vp9", "vc1", | |
| "mp3", "mp4", "m4a", "m4b", "m4v", "m2ts", "3gp", "ac3", "eac3", | |
| "bd25", "bd50", "dvd5", "dvd9", "win32", "win64", "x64", "x86", | |
| "8yo", "9yo", "10yo", "11yo", "12yo", "13yo", "14yo", "15yo", "16yo", "17yo", | |
| }) | |
| def normalize_text(text: str) -> str: | |
| """Normalize text through unicode decomposition, leetspeak mapping, and token cleaning. | |
| Steps: | |
| 1. Unicode NFKD decomposition + strip non-spacing diacritic marks. | |
| 2. Strip zero-width & bidirectional formatting characters. | |
| 3. Lowercase & map confusable Cyrillic/Greek homoglyphs. | |
| 4. Normalize punctuation and delimiters to single spaces. | |
| 5. Map leetspeak characters on mixed/obfuscated tokens while preserving pure numbers | |
| (so file numbers like 'CSS/43' never become 'cs ae') and age tokens ('14yo'). | |
| 6. Collapse repetitive runs of 3+ identical characters. | |
| """ | |
| if not text: | |
| return "" | |
| # 1. Strip Unicode tag characters, variation selectors, and format/control characters pre- and post-NFKD | |
| pre_cleaned = "".join( | |
| c for c in text | |
| if unicodedata.category(c) not in ("Cf", "Mn", "Me", "Cc") | |
| and not (0xFE00 <= ord(c) <= 0xFE0F) | |
| and not (0xE0000 <= ord(c) <= 0xE01EF) | |
| ) | |
| decomp = unicodedata.normalize("NFKD", pre_cleaned) | |
| stripped = "".join( | |
| c for c in decomp | |
| if not unicodedata.combining(c) | |
| and unicodedata.category(c) not in ("Cf", "Mn", "Me", "Cc") | |
| and not (0xFE00 <= ord(c) <= 0xFE0F) | |
| and not (0xE0000 <= ord(c) <= 0xE01EF) | |
| ) | |
| # 2. Strip zero-width characters & map homoglyphs | |
| cleaned = _ZERO_WIDTH_RE.sub("", stripped).lower().translate(_HOMOGLYPH_TRANS) | |
| # 3. Punctuation (except leet symbols @$!+|) to space first so tokens are cleanly bounded | |
| spaced = re.sub(r"[\._\-\[\]\(\)\{\}\\\/\:\,\;\*\^\~\`\'\"\?\&\=]+", " ", cleaned) | |
| # 4. Token-aware leetspeak translation: pure digits ("43", "2026") and preserved media/age tokens | |
| # are kept intact; any token containing letters or leet symbols is leet-decoded. | |
| out_tokens: list[str] = [] | |
| for tok in spaced.split(): | |
| if (len(tok) >= 2 and tok.isdigit()) or tok in _PRESERVE_ALNUM_TOKENS: | |
| out_tokens.append(tok) | |
| else: | |
| decoded = tok.translate(_LEET_TRANS) | |
| collapsed = _REPEATED_CHARS_RE.sub(r"\1\1", decoded) | |
| out_tokens.append(collapsed) | |
| joined = " ".join(out_tokens) | |
| punct_normalized = _PUNCT_RE.sub(" ", joined) | |
| return " ".join(punct_normalized.split()) | |
| def sanitize_title(title: str) -> str: | |
| """Strip private tracker auth tokens, passkeys, and user credentials for complete anonymity.""" | |
| clean = _TRACKER_TAGS_RE.sub("", title) | |
| return " ".join(clean.split()).strip() | |
| # --------------------------------------------------------------------------- | |
| # Strict Pre-Ingest CSAM & Abuse Patterns | |
| # --------------------------------------------------------------------------- | |
| # Forbidden primary CSAM terms (zero-tolerance keywords and stems; [li] handles 1->i/l ambiguity) | |
| _CSAM_DIRECT_PATTERNS = [ | |
| re.compile(r"\b(c\s*s\s*a\s*m|c\s*s\s*a\s*e)\b", re.IGNORECASE), | |
| re.compile(r"\b(pedophi[li]|paedophi[li]|pedofi[li]|paedofi[li])\w*", re.IGNORECASE), | |
| re.compile(r"\b(chi[li]d\s*porn|cp\s*vids?|cp\s*[li]inks?|cp\s*megas?|cp\s*pack\s*co[li]+ection|chi[li]d\s*erotic|underage\s*porn)\b", re.IGNORECASE), | |
| re.compile(r"\b(pre\s*teen\s*(sex|nude|porn|erotic|hardcore)|preteen[_\-\s]*(sex|nude|porn))\b", re.IGNORECASE), | |
| re.compile(r"\b([li]o[li]ita\s*(hardcore|hentai|sex|porn|nude|uncensored))\b", re.IGNORECASE), | |
| re.compile(r"\b(shotacon|shota\s*(sex|porn|hentai|hardcore))\b", re.IGNORECASE), | |
| re.compile(r"\b(todd[li]er|kindergarten|infant|e[li]ementary)\s*(sex|nude|erotic|porn)\b", re.IGNORECASE), | |
| re.compile(r"\b(jai[li]\s*bait|bare[li]y\s*[li]ega[li]\s*(underage|minor|chi[li]d))\b", re.IGNORECASE), | |
| re.compile(r"\b(incest\s*(chi[li]d|daughter|son|minor|underage))\b", re.IGNORECASE), | |
| re.compile(r"\b(?:[89]|1[0-7])\s*y\s*o\s*(?:fuck\w*|sex|porn|nude|erotic|hardcore|anal|creampie)\b", re.IGNORECASE), | |
| ] | |
| # Non-consensual exploitation & sexual violence | |
| _NONCONSENSUAL_PATTERNS = [ | |
| re.compile(r"\b(non\s*consensua[li]|rape\s*(porn|video|scene)|gang\s*rape|forced\s*sex)\b", re.IGNORECASE), | |
| re.compile(r"\b(revenge\s*porn|spycam\s*(bathroom|shower|toi[li]et|upskirt))\b", re.IGNORECASE), | |
| re.compile(r"\b(b[li]ackmai[li]\s*sex|sextortion)\b", re.IGNORECASE), | |
| ] | |
| # Extreme violence, snuff, execution, gore | |
| _EXTREME_VIOLENCE_PATTERNS = [ | |
| re.compile(r"\b(snuff\s*(fi[li]m|video|movie)|rea[li]\s*beheading|rea[li]\s*execution)\b", re.IGNORECASE), | |
| re.compile(r"\b([li]ive\s*suicide\s*(video|stream)|massacre\s*[li]ivestream)\b", re.IGNORECASE), | |
| re.compile(r"\b(rea[li]\s*torture\s*(video|fi[li]m)|crush\s*fetish\s*(anima[li]|ki[li]+))\b", re.IGNORECASE), | |
| ] | |
| # High-risk dropper extensions that are NEVER legitimate standalone software/game/media torrents | |
| _ALWAYS_MALWARE_EXTENSIONS = { | |
| ".scr", ".pif", ".hta", ".vbs", ".vbe", ".wsf", ".wsh", | |
| ".cpl", ".gadget", ".application", | |
| ".appref-ms", ".library-ms", ".search-ms", ".settingcontent-ms", ".diagcab", | |
| } | |
| # Suspicious executable & dropper extensions that must never appear disguised as media | |
| _MALWARE_EXTENSIONS = { | |
| ".exe", ".scr", ".bat", ".cmd", ".vbs", ".vbe", ".js", ".jse", | |
| ".wsf", ".wsh", ".ps1", ".psm1", ".pif", ".application", ".gadget", | |
| ".msi", ".msp", ".com", ".hta", ".cpl", ".msc", ".jar", ".lnk", ".dll", | |
| ".sys", ".drv", ".ocx", ".reg", ".url", | |
| ".chm", ".one", ".xll", ".xlam", ".docm", ".xlsm", | |
| ".appref-ms", ".library-ms", ".search-ms", ".settingcontent-ms", ".diagcab", | |
| ".pkg", ".command", | |
| } | |
| _SAMPLE_SUFFIX_RE = re.compile(r"\.sample\d*$", re.IGNORECASE) | |
| _RTLO_SPOOF_RE = re.compile(r"[\u202d\u202e]") | |
| _BDMV_JAR_PATH_RE = re.compile(r"(?:^|/)bdmv/(?:backup/)?jar/[^/]+\.jar$", re.IGNORECASE) | |
| _RENPY_SAY_VBS_CG_RE = re.compile(r"(?i)(?:^|/)lib/(?:py\d-)?windows-[^/]+/say\.vbs$") | |
| _MEDIA_CODEC_LURE_RE = re.compile( | |
| r"(?i)(?:^|(?<=[^a-z0-9]))(?:" | |
| r"codec[\s._-]*pack|" | |
| r"xvid[\s._-]*codec|" | |
| r"x264[\s._-]*video[\s._-]*codec" | |
| r")(?:$|(?=[^a-z0-9]))" | |
| ) | |
| _MALWARE_LURE_RE = re.compile( | |
| r"(?i)(?:^|(?<=[^a-z0-9]))(?:" | |
| r"required[\s._-]*to[\s._-]*play|" | |
| r"install[\s._-]*codec|" | |
| r"nitro[\s._-]*generator|" | |
| r"free[\s._-]*nitro|" | |
| r"fake[\s._-]*seeder[\s._-]*swarm[\s._-]*trap|" | |
| r"adware[\s._-]*bundle[\s._-]*installer|" | |
| r"password[\s._-]*protected[\s._-]*trap|" | |
| r"password[\s._-]*in[\s._-]*description|" | |
| r"corrupted[\s._-]*dead[\s._-]*sample|" | |
| r"corrupt\.part|" | |
| r"spam[\s._-]*link[\s._-]*redirect|" | |
| r"watch[\s._-]*full[\s._-]*movie\.url|" | |
| r"activar[\s._-]*sonido|" | |
| r"\d+(?:\.\d+)?\s*tb\s+collection\.torrent" | |
| r")(?:$|(?=[^a-z0-9]))" | |
| ) | |
| _BENIGN_BOOKMARK_URL_RE = re.compile( | |
| r"(?i)^(?:torrent[\s._-]*downloaded[\s._-]*from|downloaded[\s._-]*from|visit[\s._-]*us|rarbg|1337x|yts|tgx|torrentgalaxy|rutracker|nyaa)" | |
| ) | |
| _PASSWORD_COMPANION_RE = re.compile( | |
| r"(?i)(?:password|contrase[nñ]a|parola|mot[\s._-]*de[\s._-]*passe|passwort|how[\s._-]*to[\s._-]*extract|readme[\s._-]*pass)" | |
| ) | |
| _ANIME_OR_EP_ZIP_CG_RE = re.compile( | |
| r"(?i)(?:^\[[^\[\]]{2,30}\].*?(?:\s-\s*\d{1,3}\b|\bS\d{1,2}E\d{1,3}\b).*?\b(?:1080p|720p|2160p|480p)\b|" | |
| r"\b(?:S\d{1,2}[._ -]?E\d{1,3}|\d{1,2}x\d{2})\b.*?\b(?:1080p|720p|2160p|480p|WEB-DL|WEBRip|BluRay|BDRip|HDTV|x264|x265|HEVC)\b|" | |
| r"\s-\s\d{1,3}\s*(?:\(|\[).*?\b(?:1080p|720p|2160p|480p)\b)" | |
| ) | |
| _VIDEO_RIP_CLAIM_CG_RE = re.compile( | |
| r"(?i)(?:" | |
| r"\b(?:1080p|720p|2160p|480p|576p|4k)\b.*?\b(?:web-?dl|webrip|bluray|bdrip|brrip|hdrip|hdtv|hdcam|camrip|dvdrip|x264|x265|h\.?264|h\.?265|hevc|av1|xvid|divx)\b|" | |
| r"\b(?:web-?dl|webrip|bdrip|brrip|hdrip|hdtv|hdcam|camrip|dvdrip)\b(?:.*?\b(?:x264|x265|h\.?264|h\.?265|hevc|av1|xvid|divx|aac|ac3|ddp?5\.?1))?|" | |
| r"\b(?:19\d\d|20\d\d|s\d{1,2}[._ -]?e\d{1,3}|\d{1,2}x\d{2})\b.*?\b(?:x264|x265|h\.?264|h\.?265|hevc|av1|xvid|divx)\b" | |
| r")" | |
| ) | |
| _OPTICAL_DISC_TITLE_CG_RE = re.compile( | |
| r"(?i)\b(?:dvd(?:[59]|r|-r|-rom)?|bd(?:25|50|66|100|mv|rom|-r)|blu-?ray[\s._-]*disc|full[\s._-]*bd|untouched|remux|pal|ntsc|iso|udf|sacd|dvd-?audio|3d[\s._-]*iso)\b" | |
| ) | |
| _BENIGN_ZIP_CG_RE = re.compile( | |
| r"(?i)\b(?:batch|complete|season[ ._-]*\d+|s\d{1,2}[ ._-]*complete|e?\d{1,3}[ ._-]*[~-][ ._-]*e?\d{1,3}|시즌\s*\d+|완결|全集|合集|" | |
| r"ost|soundtrack|flac|mp3|wav|lossless|album|discography|artbook|scans?|wallpapers?|fonts?|subs?|subtitles?|" | |
| r"cbz|cbr|comic|manga|magazine|ebook|epub|pdf|audiobook|m4b|templates?|presets?|samples?|stems)\b" | |
| ) | |
| _SINGLE_ARCHIVE_TRAP_EXTS = {".zip", ".zipx", ".rar", ".7z", ".ace", ".cab"} | |
| _DISK_IMAGE_TRAP_EXTS = {".iso", ".img", ".vhd", ".vhdx", ".udf", ".dmg"} | |
| _VIDEO_MEDIA_EXTS = {".mp4", ".mkv", ".avi", ".mov", ".wmv", ".webm", ".flv", ".m4v", ".mpg", ".mpeg", ".ts", ".m2ts", ".vob", ".ogv", ".3gp"} | |
| _BENIGN_COURSE_EXTS = _VIDEO_MEDIA_EXTS | { | |
| ".mp3", ".flac", ".m4a", ".m4b", ".aac", ".ogg", ".wav", | |
| ".pdf", ".epub", ".mobi", ".djvu", ".cbz", ".cbr", | |
| ".srt", ".ass", ".ssa", ".vtt", ".sub", | |
| ".gp3", ".gp4", ".gp5", ".gpx", ".ptb", ".ipynb", ".md", ".html", | |
| } | |
| _ARCHIVE_EXTENSIONS = {".zip", ".zipx", ".rar", ".7z", ".tar", ".gz", ".bz2", ".xz", ".iso", ".img"} | |
| _DISGUISED_MEDIA_STEMS = ( | |
| ".mp4.", ".mkv.", ".avi.", ".mov.", ".wmv.", ".webm.", ".flv.", ".m4v.", | |
| ".mp4 ", ".mkv ", ".avi ", ".mov ", | |
| ".mp3.", ".flac.", ".wav.", ".m4a.", ".ogg.", ".aac.", ".opus.", | |
| ".pdf.", ".epub.", ".mobi.", ".cbz.", ".cbr.", ".djvu.", | |
| ".jpg.", ".jpeg.", ".png.", ".gif.", ".webp.", ".srt.", ".ass.", ".sub.", | |
| ) | |
| def _clean_filename_for_ext(raw_name: str) -> str: | |
| s = raw_name.lower().strip().rstrip(". \t\r\n\x00") | |
| s = _SAMPLE_SUFFIX_RE.sub("", s).rstrip(". \t\r\n\x00") | |
| return s | |
| def _match_malware_ext(clean_lower: str) -> str | None: | |
| dot = clean_lower.rfind(".") | |
| if dot == -1: | |
| return None | |
| ext = clean_lower[dot:] | |
| return ext if ext in _MALWARE_EXTENSIONS else None | |
| def _is_disk_image_mount_trap_cg(title_raw: str) -> bool: | |
| stem = re.sub(r"\.(?:iso|img|vhd|vhdx|udf|dmg)$", "", title_raw.strip(), flags=re.IGNORECASE) | |
| if _OPTICAL_DISC_TITLE_CG_RE.search(stem) or _BENIGN_ZIP_CG_RE.search(stem): | |
| return False | |
| return bool(_VIDEO_RIP_CLAIM_CG_RE.search(stem) or _ANIME_OR_EP_ZIP_CG_RE.search(stem)) | |
| # --------------------------------------------------------------------------- | |
| # CSAM & Abuse Guard Implementation | |
| # --------------------------------------------------------------------------- | |
| class CSAMGuard: | |
| """Comprehensive content safety guard enforcing CSAM, abuse, and malware blocking.""" | |
| def __init__(self, tombstone_store: TombstoneStore | None = None) -> None: | |
| self.tombstone_store = tombstone_store or TombstoneStore() | |
| def inspect_text(self, text: str) -> ContentFilterResult: | |
| """Inspect a single string (title, file path, tag) for CSAM and harmful content.""" | |
| if not text: | |
| return ContentFilterResult( | |
| is_safe=True, | |
| violation_type=None, | |
| reason=None, | |
| matched_pattern=None, | |
| risk_score=0.0, | |
| sanitized_title="", | |
| ) | |
| sanitized = sanitize_title(text) | |
| normalized = normalize_text(sanitized) | |
| # Preserve standalone "css" (Cascading Style Sheets) before collapsing single-char runs or stretching | |
| norm_css_safe = re.sub(r"\bcss\b", "css_style_token", normalized) | |
| de_spaced = _SINGLE_CHAR_RUN_RE.sub(r"\1", norm_css_safe) | |
| # Collapse arbitrary character stretching while preserving valid English double letters in target stems | |
| de_stretched = re.sub(r"(.)\1+", r"\1", de_spaced).replace("cs_style_token", "css") | |
| de_spaced = de_spaced.replace("css_style_token", "css") | |
| for raw_stem, canonical_stem in ( | |
| ("ten", "teen"), | |
| ("todler", "toddler"), | |
| ("snuf", "snuff"), | |
| ("masacre", "massacre"), | |
| ("kil", "kill"), | |
| ("bathrom", "bathroom"), | |
| ): | |
| de_stretched = de_stretched.replace(raw_stem, canonical_stem) | |
| candidates = tuple(dict.fromkeys((normalized, de_spaced, de_stretched))) | |
| for cand in candidates: | |
| # 1. Direct CSAM patterns | |
| for pat in _CSAM_DIRECT_PATTERNS: | |
| m = pat.search(cand) | |
| if m: | |
| return ContentFilterResult( | |
| is_safe=False, | |
| violation_type=CSAMViolationType.CSAM_EXPLOITATION, | |
| reason=f"Matched direct CSAM pattern: '{m.group()}'", | |
| matched_pattern=pat.pattern, | |
| risk_score=1.0, | |
| sanitized_title=sanitized, | |
| ) | |
| # 2. Non-consensual exploitation patterns | |
| for pat in _NONCONSENSUAL_PATTERNS: | |
| m = pat.search(cand) | |
| if m: | |
| return ContentFilterResult( | |
| is_safe=False, | |
| violation_type=CSAMViolationType.NONCONSENSUAL_ABUSE, | |
| reason=f"Matched non-consensual exploitation pattern: '{m.group()}'", | |
| matched_pattern=pat.pattern, | |
| risk_score=0.98, | |
| sanitized_title=sanitized, | |
| ) | |
| # 3. Extreme violence & gore patterns | |
| for pat in _EXTREME_VIOLENCE_PATTERNS: | |
| m = pat.search(cand) | |
| if m: | |
| return ContentFilterResult( | |
| is_safe=False, | |
| violation_type=CSAMViolationType.EXTREME_VIOLENCE_GORE, | |
| reason=f"Matched extreme violence pattern: '{m.group()}'", | |
| matched_pattern=pat.pattern, | |
| risk_score=0.95, | |
| sanitized_title=sanitized, | |
| ) | |
| return ContentFilterResult( | |
| is_safe=True, | |
| violation_type=None, | |
| reason=None, | |
| matched_pattern=None, | |
| risk_score=0.0, | |
| sanitized_title=sanitized, | |
| ) | |
| def inspect_torrent( | |
| self, | |
| infohash: str, | |
| title: str, | |
| file_list: Sequence[str | dict[str, Any]] | None = None, | |
| tags: Sequence[str] | None = None, | |
| category: str | None = None, | |
| ) -> ContentFilterResult: | |
| """Structural pre-ingest inspection over entire torrent metadata. | |
| Verifies: | |
| 1. Infohash is not on the Ed25519-signed blinded tombstone purge list. | |
| 2. Torrent title contains no CSAM/abuse keywords or passkey leakage. | |
| 3. None of the internal files contain CSAM tokens. | |
| 4. No disguised executable payloads, RTLO spoofing, mount traps, or archive traps exist. | |
| 5. Tags contain no prohibited content. | |
| """ | |
| clean_infohash = str(infohash).strip().lower() | |
| if not re.fullmatch(r"[0-9a-f]{40}", clean_infohash): | |
| return ContentFilterResult( | |
| is_safe=False, | |
| violation_type=CSAMViolationType.MALWARE_PAYLOAD, | |
| reason="Malformed non-hexadecimal 40-char infohash rejected", | |
| matched_pattern=clean_infohash[:40], | |
| risk_score=1.0, | |
| sanitized_title=sanitize_title(title), | |
| ) | |
| # 1. Tombstone purge check | |
| if self.tombstone_store.is_tombstoned(clean_infohash): | |
| return ContentFilterResult( | |
| is_safe=False, | |
| violation_type=CSAMViolationType.TOMBSTONED_HASH, | |
| reason=f"Infohash {clean_infohash} is present on the signed revocation tombstone list", | |
| matched_pattern=clean_infohash, | |
| risk_score=1.0, | |
| sanitized_title=sanitize_title(title), | |
| ) | |
| # 2. Inspect title for CSAM / non-consensual / extreme violence | |
| res = self.inspect_text(title) | |
| if not res.is_safe: | |
| return res | |
| # 3. Inspect tags | |
| if tags: | |
| for tag in tags: | |
| tag_res = self.inspect_text(tag) | |
| if not tag_res.is_safe: | |
| return tag_res | |
| # 4. Inspect file list text for CSAM / abuse first (preserves violation_type priority) | |
| parsed_files: list[tuple[str, str, int]] = [] | |
| if file_list: | |
| for f in file_list: | |
| fname = f if isinstance(f, str) else str(f.get("path") or f.get("p") or f.get("name") or "") | |
| fsize = int((f.get("size") or f.get("s") or 0) if isinstance(f, dict) else 0) | |
| f_res = self.inspect_text(fname) | |
| if not f_res.is_safe: | |
| return f_res | |
| parsed_files.append((fname, _clean_filename_for_ext(fname), fsize)) | |
| # 5. Check Unicode Right-to-Left Override (RTLO) extension spoofing in title or files | |
| if _RTLO_SPOOF_RE.search(title) or any(_RTLO_SPOOF_RE.search(fn) for fn, _, _ in parsed_files): | |
| return ContentFilterResult( | |
| is_safe=False, | |
| violation_type=CSAMViolationType.MALWARE_PAYLOAD, | |
| reason=f"Blocked Unicode RTLO extension spoofing character in torrent: '{title}'", | |
| matched_pattern="U+202E_RTLO", | |
| risk_score=1.0, | |
| sanitized_title=res.sanitized_title, | |
| ) | |
| title_clean = _clean_filename_for_ext(title) | |
| # 6. Universal malware/honeypot lure patterns (blocked across ALL categories) | |
| m_lure = _MALWARE_LURE_RE.search(title_clean) | |
| if m_lure: | |
| return ContentFilterResult( | |
| is_safe=False, | |
| violation_type=CSAMViolationType.MALWARE_PAYLOAD, | |
| reason=f"Blocked explicit malware/honeypot lure in title: '{title}'", | |
| matched_pattern=m_lure.group(0), | |
| risk_score=0.99, | |
| sanitized_title=res.sanitized_title, | |
| ) | |
| for fname, f_lower, _ in parsed_files: | |
| m_flure = _MALWARE_LURE_RE.search(f_lower) | |
| if m_flure: | |
| return ContentFilterResult( | |
| is_safe=False, | |
| violation_type=CSAMViolationType.MALWARE_PAYLOAD, | |
| reason=f"Blocked explicit malware/honeypot lure in file: '{fname}'", | |
| matched_pattern=m_flure.group(0), | |
| risk_score=0.99, | |
| sanitized_title=res.sanitized_title, | |
| ) | |
| title_claims_video = bool(_ANIME_OR_EP_ZIP_CG_RE.search(title_clean) or _VIDEO_RIP_CLAIM_CG_RE.search(title_clean)) | |
| # 7. Title-level disguised media double-extension & unconditional dropper extension checks | |
| t_ext = _match_malware_ext(title_clean) | |
| if t_ext is not None: | |
| if ( | |
| any(stem in title_clean for stem in _DISGUISED_MEDIA_STEMS) | |
| or t_ext in _ALWAYS_MALWARE_EXTENSIONS | |
| or (title_claims_video and t_ext in (".one", ".xll", ".xlam")) | |
| ): | |
| return ContentFilterResult( | |
| is_safe=False, | |
| violation_type=CSAMViolationType.MALWARE_PAYLOAD, | |
| reason=f"Disguised executable or high-risk dropper in torrent title: '{title}'", | |
| matched_pattern=t_ext, | |
| risk_score=0.99, | |
| sanitized_title=res.sanitized_title, | |
| ) | |
| cat_lower = (category or "").lower().strip() | |
| is_strict_media_category = cat_lower in { | |
| "movie", "tv_show", "music", "anime", "book", "ebook", "audiobook", "comic", "xxx" | |
| } | |
| is_course_category = cat_lower == "course" | |
| has_large_real_video = False | |
| has_any_video = False | |
| has_bdmv_structure = "bdmv" in title_clean | |
| has_renpy_structure = False | |
| benign_course_files = 0 | |
| benign_course_bytes = 0 | |
| has_password_companion = False | |
| image_file_count = 0 | |
| image_total_bytes = 0 | |
| for _, f_lower, fsize in parsed_files: | |
| f_posix = f_lower.replace("\\", "/") | |
| if "/bdmv/" in f_posix or f_posix.startswith("bdmv/") or f_posix.endswith((".m2ts", ".bdmv", ".mpls")): | |
| has_bdmv_structure = True | |
| if f_posix.endswith((".rpa", ".rpyb")) and fsize >= 65536: | |
| has_renpy_structure = True | |
| if _PASSWORD_COMPANION_RE.search(f_posix): | |
| has_password_companion = True | |
| dot = f_posix.rfind(".") | |
| f_ext = f_posix[dot:] if dot != -1 else "" | |
| if f_ext in _BENIGN_COURSE_EXTS and fsize >= 65536: | |
| benign_course_files += 1 | |
| benign_course_bytes += fsize | |
| if f_ext in (".jpg", ".jpeg", ".png", ".webp") and fsize > 0: | |
| image_file_count += 1 | |
| image_total_bytes += fsize | |
| if f_ext in _VIDEO_MEDIA_EXTS: | |
| has_any_video = True | |
| if fsize >= 100_000_000: | |
| has_large_real_video = True | |
| has_real_image_set = image_file_count >= 5 and image_total_bytes >= 5_000_000 | |
| has_real_course_media = benign_course_files >= 3 and benign_course_bytes >= 50_000_000 | |
| # Multi-lesson courses (>= 5 files, >= 3 sized lesson files, >= 50 MB total lesson media) legitimately | |
| # include .js, .ps1, .bat, or interactive player .exe files; zero-byte or tiny decoys are still checked. | |
| check_media_executables = is_strict_media_category or ( | |
| is_course_category and (len(parsed_files) < 5 or not has_real_course_media) | |
| ) | |
| # Contextual codec-pack lure check: blocks codec-pack installers disguised as or bundled with media | |
| # while sparing legitimate standalone codec-pack software utilities. | |
| if check_media_executables or has_any_video or title_claims_video: | |
| m_clure = _MEDIA_CODEC_LURE_RE.search(title_clean) | |
| if m_clure and (has_any_video or title_claims_video): | |
| return ContentFilterResult( | |
| is_safe=False, | |
| violation_type=CSAMViolationType.MALWARE_PAYLOAD, | |
| reason=f"Blocked media codec-pack lure in title: '{title}'", | |
| matched_pattern=m_clure.group(0), | |
| risk_score=0.99, | |
| sanitized_title=res.sanitized_title, | |
| ) | |
| for fname, f_lower, _ in parsed_files: | |
| m_fclure = _MEDIA_CODEC_LURE_RE.search(f_lower) | |
| if m_fclure: | |
| return ContentFilterResult( | |
| is_safe=False, | |
| violation_type=CSAMViolationType.MALWARE_PAYLOAD, | |
| reason=f"Blocked explicit malware/honeypot lure in file: '{fname}'", | |
| matched_pattern=m_fclure.group(0), | |
| risk_score=0.99, | |
| sanitized_title=res.sanitized_title, | |
| ) | |
| # 8. Cold / bare-title checks when file_list is empty | |
| if not parsed_files and (check_media_executables or cat_lower in ("unknown", "")): | |
| if t_ext is not None and not _BENIGN_ZIP_CG_RE.search(title_clean): | |
| return ContentFilterResult( | |
| is_safe=False, | |
| violation_type=CSAMViolationType.MALWARE_PAYLOAD, | |
| reason=f"Disguised executable or script extension in media title: '{title}'", | |
| matched_pattern=t_ext, | |
| risk_score=0.99, | |
| sanitized_title=res.sanitized_title, | |
| ) | |
| dot_t = title_clean.rfind(".") | |
| bare_ext = title_clean[dot_t:] if dot_t != -1 else "" | |
| if bare_ext in _SINGLE_ARCHIVE_TRAP_EXTS and not _BENIGN_ZIP_CG_RE.search(title_clean): | |
| if title_claims_video: | |
| return ContentFilterResult( | |
| is_safe=False, | |
| violation_type=CSAMViolationType.MALWARE_PAYLOAD, | |
| reason=f"Single-archive media malware trap in title: '{title}'", | |
| matched_pattern=bare_ext, | |
| risk_score=0.99, | |
| sanitized_title=res.sanitized_title, | |
| ) | |
| if bare_ext in _DISK_IMAGE_TRAP_EXTS and ( | |
| any(stem in title_clean for stem in _DISGUISED_MEDIA_STEMS) or _is_disk_image_mount_trap_cg(title_clean) | |
| ): | |
| return ContentFilterResult( | |
| is_safe=False, | |
| violation_type=CSAMViolationType.MALWARE_PAYLOAD, | |
| reason=f"Disk image mount malware trap ({bare_ext}) in media title: '{title}'", | |
| matched_pattern=bare_ext, | |
| risk_score=0.99, | |
| sanitized_title=res.sanitized_title, | |
| ) | |
| # 9. File-list malware, mount trap, and archive trap inspection | |
| if parsed_files: | |
| for fname, f_lower, fsize in parsed_files: | |
| f_posix = f_lower.replace("\\", "/") | |
| base_lower = f_posix.rsplit("/", 1)[-1] | |
| if ( | |
| base_lower in ("do_not_mirror.exe", "rarbg_do_not_mirror.exe", "rarbg.com_do_not_mirror.exe") | |
| and (0 < fsize <= 512) | |
| and (has_large_real_video or has_real_image_set) | |
| ): | |
| continue | |
| if ( | |
| _RENPY_SAY_VBS_CG_RE.search(f_posix) | |
| and not is_strict_media_category | |
| and not title_claims_video | |
| and not has_any_video | |
| and has_renpy_structure | |
| and (0 < fsize <= 4096) | |
| ): | |
| continue | |
| is_disguised = any(stem in f_lower for stem in _DISGUISED_MEDIA_STEMS) | |
| dot_f = f_lower.rfind(".") | |
| raw_ext = f_lower[dot_f:] if dot_f != -1 else "" | |
| if is_disguised and raw_ext in _DISK_IMAGE_TRAP_EXTS: | |
| return ContentFilterResult( | |
| is_safe=False, | |
| violation_type=CSAMViolationType.MALWARE_PAYLOAD, | |
| reason=f"Disguised disk-image payload found in torrent: '{fname}'", | |
| matched_pattern=raw_ext, | |
| risk_score=0.99, | |
| sanitized_title=res.sanitized_title, | |
| ) | |
| ext = _match_malware_ext(f_lower) | |
| if ext is not None: | |
| # Exempt legitimate Blu-ray BD-J Java menu JARs (BDMV/JAR/00000.jar) | |
| if ext == ".jar" and has_bdmv_structure and _BDMV_JAR_PATH_RE.search(f_posix): | |
| continue | |
| # Exempt known small tracker bookmark (.url) only when not disguised and real media is present | |
| if ( | |
| ext == ".url" | |
| and not is_disguised | |
| and (0 < fsize <= 2048) | |
| and _BENIGN_BOOKMARK_URL_RE.search(base_lower) | |
| and (has_large_real_video or has_real_image_set or has_real_course_media) | |
| ): | |
| continue | |
| if ( | |
| check_media_executables | |
| or is_disguised | |
| or ext in _ALWAYS_MALWARE_EXTENSIONS | |
| or (title_claims_video and ext in (".one", ".xll", ".xlam")) | |
| ): | |
| return ContentFilterResult( | |
| is_safe=False, | |
| violation_type=CSAMViolationType.MALWARE_PAYLOAD, | |
| reason=f"Disguised executable or dropper found in media torrent: '{fname}'", | |
| matched_pattern=ext, | |
| risk_score=0.99, | |
| sanitized_title=res.sanitized_title, | |
| ) | |
| if check_media_executables and not has_any_video and len(parsed_files) <= 3: | |
| # Optical / Disk Image Mount Trap (.iso, .img, .vhd, .vhdx, .udf, .dmg) | |
| for fname, f_lower, _ in parsed_files: | |
| dot = f_lower.rfind(".") | |
| f_ext = f_lower[dot:] if dot != -1 else "" | |
| if f_ext in _DISK_IMAGE_TRAP_EXTS and _is_disk_image_mount_trap_cg(title_clean): | |
| return ContentFilterResult( | |
| is_safe=False, | |
| violation_type=CSAMViolationType.MALWARE_PAYLOAD, | |
| reason=f"Disk image mount malware trap ({f_ext}) in stream/rip-claiming media torrent: '{fname}'", | |
| matched_pattern=f_ext, | |
| risk_score=0.99, | |
| sanitized_title=res.sanitized_title, | |
| ) | |
| # Single-archive media trap (.zip, .zipx, .rar, .7z, .ace, .cab) | |
| if not _BENIGN_ZIP_CG_RE.search(title_clean): | |
| arch_matches: list[tuple[str, str]] = [] | |
| for fname, f_lower, _ in parsed_files: | |
| dot = f_lower.rfind(".") | |
| f_ext = f_lower[dot:] if dot != -1 else "" | |
| if f_ext in _SINGLE_ARCHIVE_TRAP_EXTS: | |
| arch_matches.append((fname, f_ext)) | |
| if len(arch_matches) == 1: | |
| arch_fname, arch_ext = arch_matches[0] | |
| if has_password_companion or ( | |
| len(parsed_files) <= 2 | |
| and cat_lower in ("movie", "tv_show", "anime", "unknown", "") | |
| and (_ANIME_OR_EP_ZIP_CG_RE.search(title_clean) or _VIDEO_RIP_CLAIM_CG_RE.search(title_clean)) | |
| ): | |
| return ContentFilterResult( | |
| is_safe=False, | |
| violation_type=CSAMViolationType.MALWARE_PAYLOAD, | |
| reason=f"Single-archive episode/movie malware trap in media torrent: '{arch_fname}'", | |
| matched_pattern=arch_ext, | |
| risk_score=0.99, | |
| sanitized_title=res.sanitized_title, | |
| ) | |
| return ContentFilterResult( | |
| is_safe=True, | |
| violation_type=None, | |
| reason=None, | |
| matched_pattern=None, | |
| risk_score=0.0, | |
| sanitized_title=res.sanitized_title, | |
| ) | |
| # --------------------------------------------------------------------------- | |
| # Armiger Conformal Safety Gate Interface | |
| # --------------------------------------------------------------------------- | |
| class ArmigerSafetyGate: | |
| """Evaluates Mondrian conformal thresholds to prevent malware injection and adult leakage.""" | |
| def __init__( | |
| self, | |
| adult_threshold: float = 0.5451, | |
| junk_threshold: float = 0.1010, | |
| scope_threshold: float = 0.4770, | |
| ) -> None: | |
| self.adult_threshold = adult_threshold | |
| self.junk_threshold = junk_threshold | |
| self.scope_threshold = scope_threshold | |
| def evaluate( | |
| self, | |
| p_adult: float, | |
| p_junk: float, | |
| p_excluded_scope: float, | |
| target_category: str, | |
| ) -> tuple[bool, str, CSAMViolationType | None]: | |
| """Validate safety gates for a torrent candidate. | |
| Rules: | |
| - If p_junk >= junk_threshold: Drop as malware/phishing trap. | |
| - If p_excluded_scope >= scope_threshold: Drop as out-of-scope. | |
| - If p_adult >= adult_threshold: Must ONLY be routed to 'xxx' category. | |
| If target_category is not 'xxx', reject to maintain strict mainstream isolation. | |
| """ | |
| for pname, pval in (("p_adult", p_adult), ("p_junk", p_junk), ("p_excluded_scope", p_excluded_scope)): | |
| if isinstance(pval, bool) or not isinstance(pval, (int, float)) or not math.isfinite(pval) or pval < 0.0 or pval > 1.0: | |
| return False, f"invalid_probability_{pname}:{pval}", CSAMViolationType.SAFETY_GATE_JUNK | |
| if p_junk >= self.junk_threshold: | |
| return False, f"junk_probability_exceeded:{p_junk:.4f}>={self.junk_threshold:.4f}", CSAMViolationType.SAFETY_GATE_JUNK | |
| if p_excluded_scope >= self.scope_threshold: | |
| return False, f"out_of_scope_probability_exceeded:{p_excluded_scope:.4f}>={self.scope_threshold:.4f}", CSAMViolationType.SAFETY_GATE_OUT_OF_SCOPE | |
| is_adult = bool(p_adult >= self.adult_threshold) | |
| if is_adult and target_category != "xxx": | |
| return False, f"unisolated_adult_leakage:{p_adult:.4f}>={self.adult_threshold:.4f}_into_{target_category}", CSAMViolationType.UNISOLATED_ADULT | |
| if not is_adult and target_category == "xxx": | |
| # Target is xxx but confidence of adult is below threshold; allow or flag | |
| pass | |
| return True, "passed", None | |
| # --------------------------------------------------------------------------- | |
| # Post-Quantum Primitives (ML-KEM-768 FIPS 203 & WOTS+-SHAKE256 FIPS 205/RFC 8391) | |
| # --------------------------------------------------------------------------- | |
| class MLKEM768: | |
| """Self-contained NIST FIPS 203 ML-KEM-768 (Module-LWE Post-Quantum KEM) implementation.""" | |
| Q = 3329 | |
| N = 256 | |
| K = 3 | |
| EK_BYTES = 1184 | |
| DK_BYTES = 2400 | |
| CT_BYTES = 1088 | |
| SS_BYTES = 32 | |
| ZETAS = ( | |
| 1, 1729, 2580, 3289, 2642, 630, 1897, 848, 1062, 1919, 193, 797, 2786, 3260, 569, 1746, | |
| 296, 2447, 1339, 1476, 3046, 56, 2240, 1333, 1426, 2094, 535, 2882, 2393, 2879, 1974, 821, | |
| 289, 331, 3253, 1756, 1197, 2304, 2277, 2055, 650, 1977, 2513, 632, 2865, 33, 1320, 1915, | |
| 2319, 1435, 807, 452, 1438, 2868, 1534, 2402, 2647, 2617, 1481, 648, 2474, 3110, 1227, 910, | |
| 17, 2761, 583, 2649, 1637, 723, 2288, 1100, 1409, 2662, 3281, 233, 756, 2156, 3015, 3050, | |
| 1703, 1651, 2789, 1789, 1847, 952, 1461, 2687, 939, 2308, 2437, 2388, 733, 2337, 268, 641, | |
| 1584, 2298, 2037, 3220, 375, 2549, 2090, 1645, 1063, 319, 2773, 757, 2099, 561, 2466, 2594, | |
| 2804, 1092, 403, 1026, 1143, 2150, 2775, 886, 1722, 1212, 1874, 1029, 2110, 2935, 885, 2154, | |
| ) | |
| GAMMAS = tuple(pow(17, 2 * int(f"{i:07b}"[::-1], 2) + 1, 3329) for i in range(128)) | |
| _CBD2_LO = tuple(((0, 1, 1, 2)[b & 3] - (0, 1, 1, 2)[(b >> 2) & 3]) % 3329 for b in range(256)) | |
| _CBD2_HI = tuple(((0, 1, 1, 2)[(b >> 4) & 3] - (0, 1, 1, 2)[(b >> 6) & 3]) % 3329 for b in range(256)) | |
| _NTT_STEPS: List[List[Tuple[int, int, int, int]]] = [] | |
| _INTT_STEPS: List[List[Tuple[int, int, int, int]]] = [] | |
| def _init_tables(cls) -> None: | |
| if cls._NTT_STEPS: | |
| return | |
| k = 1 | |
| length = 128 | |
| while length >= 2: | |
| layer = [] | |
| for start in range(0, 256, 2 * length): | |
| zeta = cls.ZETAS[k] | |
| k += 1 | |
| layer.append((zeta, start, start + length, length)) | |
| cls._NTT_STEPS.append(layer) | |
| length >>= 1 | |
| k = 127 | |
| length = 2 | |
| while length <= 128: | |
| layer = [] | |
| for start in range(0, 256, 2 * length): | |
| zeta = cls.ZETAS[k] | |
| k -= 1 | |
| layer.append((zeta, start, start + length, length)) | |
| cls._INTT_STEPS.append(layer) | |
| length <<= 1 | |
| def ntt(cls, f: List[int]) -> List[int]: | |
| p = list(f) | |
| q = 3329 | |
| for layer in cls._NTT_STEPS: | |
| for zeta, s, e, l in layer: | |
| for j in range(s, e): | |
| t = (zeta * p[j + l]) % q | |
| p[j + l] = (p[j] - t) % q | |
| p[j] = (p[j] + t) % q | |
| return p | |
| def intt(cls, f: List[int]) -> List[int]: | |
| p = list(f) | |
| q = 3329 | |
| for layer in cls._INTT_STEPS: | |
| for zeta, s, e, l in layer: | |
| for j in range(s, e): | |
| t = p[j] | |
| p[j] = (t + p[j + l]) % q | |
| p[j + l] = (zeta * (p[j + l] - t)) % q | |
| return [(x * 3303) % q for x in p] | |
| def basemul_acc(cls, a_vec: List[List[int]], b_vec: List[List[int]]) -> List[int]: | |
| res = [0] * 256 | |
| q = 3329 | |
| gammas = cls.GAMMAS | |
| for a, b in zip(a_vec, b_vec): | |
| for i in range(128): | |
| idx = i << 1 | |
| a0, a1 = a[idx], a[idx + 1] | |
| b0, b1 = b[idx], b[idx + 1] | |
| res[idx] = (res[idx] + a0 * b0 + a1 * b1 * gammas[i]) % q | |
| res[idx + 1] = (res[idx + 1] + a0 * b1 + a1 * b0) % q | |
| return res | |
| def sample_ntt(rho: bytes, j: int, i: int) -> List[int]: | |
| stream = hashlib.shake_128(rho + bytes([j, i])).digest(672) | |
| a: List[int] = [] | |
| idx = 0 | |
| while len(a) < 256: | |
| b0, b1, b2 = stream[idx], stream[idx + 1], stream[idx + 2] | |
| idx += 3 | |
| d1 = b0 + ((b1 & 0x0F) << 8) | |
| d2 = (b1 >> 4) + (b2 << 4) | |
| if d1 < 3329: | |
| a.append(d1) | |
| if d2 < 3329 and len(a) < 256: | |
| a.append(d2) | |
| return a | |
| def sample_cbd2(cls, prf_bytes: bytes) -> List[int]: | |
| f = [0] * 256 | |
| lo = cls._CBD2_LO | |
| hi = cls._CBD2_HI | |
| for i, b in enumerate(prf_bytes): | |
| idx = i << 1 | |
| f[idx] = lo[b] | |
| f[idx + 1] = hi[b] | |
| return f | |
| def encode_12(f: List[int]) -> bytes: | |
| out = bytearray(384) | |
| for i in range(128): | |
| a = f[2 * i] % 3329 | |
| b = f[2 * i + 1] % 3329 | |
| idx = 3 * i | |
| out[idx] = a & 0xFF | |
| out[idx + 1] = ((a >> 8) & 0x0F) | ((b & 0x0F) << 4) | |
| out[idx + 2] = (b >> 4) & 0xFF | |
| return bytes(out) | |
| def decode_12(buf: bytes) -> List[int]: | |
| f = [0] * 256 | |
| for i in range(128): | |
| idx = 3 * i | |
| b0, b1, b2 = buf[idx], buf[idx + 1], buf[idx + 2] | |
| f[2 * i] = (b0 + ((b1 & 0x0F) << 8)) % 3329 | |
| f[2 * i + 1] = ((b1 >> 4) + (b2 << 4)) % 3329 | |
| return f | |
| def compress_encode_10(f: List[int]) -> bytes: | |
| out = bytearray(320) | |
| for i in range(64): | |
| c0 = ((f[4 * i] << 10) + 1664) // 3329 & 0x3FF | |
| c1 = ((f[4 * i + 1] << 10) + 1664) // 3329 & 0x3FF | |
| c2 = ((f[4 * i + 2] << 10) + 1664) // 3329 & 0x3FF | |
| c3 = ((f[4 * i + 3] << 10) + 1664) // 3329 & 0x3FF | |
| idx = 5 * i | |
| out[idx] = c0 & 0xFF | |
| out[idx + 1] = ((c0 >> 8) & 0x03) | ((c1 & 0x3F) << 2) | |
| out[idx + 2] = ((c1 >> 6) & 0x0F) | ((c2 & 0x0F) << 4) | |
| out[idx + 3] = ((c2 >> 4) & 0x3F) | ((c3 & 0x03) << 6) | |
| out[idx + 4] = (c3 >> 2) & 0xFF | |
| return bytes(out) | |
| def decode_decompress_10(buf: bytes) -> List[int]: | |
| f = [0] * 256 | |
| for i in range(64): | |
| idx = 5 * i | |
| b0, b1, b2, b3, b4 = buf[idx], buf[idx + 1], buf[idx + 2], buf[idx + 3], buf[idx + 4] | |
| c0 = b0 | ((b1 & 0x03) << 8) | |
| c1 = (b1 >> 2) | ((b2 & 0x0F) << 6) | |
| c2 = (b2 >> 4) | ((b3 & 0x3F) << 4) | |
| c3 = (b3 >> 6) | (b4 << 2) | |
| f[4 * i] = (c0 * 3329 + 512) >> 10 | |
| f[4 * i + 1] = (c1 * 3329 + 512) >> 10 | |
| f[4 * i + 2] = (c2 * 3329 + 512) >> 10 | |
| f[4 * i + 3] = (c3 * 3329 + 512) >> 10 | |
| return f | |
| def compress_encode_4(f: List[int]) -> bytes: | |
| out = bytearray(128) | |
| for i in range(128): | |
| c0 = ((f[2 * i] << 4) + 1664) // 3329 & 0x0F | |
| c1 = ((f[2 * i + 1] << 4) + 1664) // 3329 & 0x0F | |
| out[i] = c0 | (c1 << 4) | |
| return bytes(out) | |
| def decode_decompress_4(buf: bytes) -> List[int]: | |
| f = [0] * 256 | |
| for i in range(128): | |
| b = buf[i] | |
| f[2 * i] = ((b & 0x0F) * 3329 + 8) >> 4 | |
| f[2 * i + 1] = (((b >> 4) & 0x0F) * 3329 + 8) >> 4 | |
| return f | |
| def encode_1(m: bytes) -> List[int]: | |
| f = [0] * 256 | |
| for i in range(32): | |
| b = m[i] | |
| base = i << 3 | |
| for j in range(8): | |
| f[base + j] = 1665 if ((b >> j) & 1) else 0 | |
| return f | |
| def decode_1(f: List[int]) -> bytes: | |
| out = bytearray(32) | |
| for i in range(32): | |
| byte_val = 0 | |
| base = i << 3 | |
| for j in range(8): | |
| c = (((f[base + j] << 1) + 1664) // 3329) & 1 | |
| byte_val |= (c << j) | |
| out[i] = byte_val | |
| return bytes(out) | |
| def _expand_ek(cls, ek: bytes) -> Tuple[List[List[int]], List[List[List[int]]], bytes]: | |
| cls._init_tables() | |
| t_hat = [cls.decode_12(ek[384 * i : 384 * (i + 1)]) for i in range(3)] | |
| rho = ek[1152:1184] | |
| a_hat_t = [[cls.sample_ntt(rho, i, j) for j in range(3)] for i in range(3)] | |
| ek_hash = hashlib.sha3_256(ek).digest() | |
| return t_hat, a_hat_t, ek_hash | |
| def _expand_dk(cls, dk: bytes) -> Tuple[List[List[int]], bytes, bytes, bytes, List[List[int]], List[List[List[int]]]]: | |
| cls._init_tables() | |
| s_hat = [cls.decode_12(dk[384 * i : 384 * (i + 1)]) for i in range(3)] | |
| ek = dk[1152:2336] | |
| h = dk[2336:2368] | |
| z = dk[2368:2400] | |
| t_hat, a_hat_t, _ = cls._expand_ek(ek) | |
| return s_hat, ek, h, z, t_hat, a_hat_t | |
| def keygen(cls, seed_d: Optional[bytes] = None, seed_z: Optional[bytes] = None) -> Tuple[bytes, bytes]: | |
| cls._init_tables() | |
| d = seed_d if seed_d is not None else os.urandom(32) | |
| z = seed_z if seed_z is not None else os.urandom(32) | |
| g = hashlib.sha3_512(d + bytes([3])).digest() | |
| rho, sigma = g[:32], g[32:] | |
| a_hat = [[cls.sample_ntt(rho, j, i) for j in range(3)] for i in range(3)] | |
| s_hat = [cls.ntt(cls.sample_cbd2(hashlib.shake_256(sigma + bytes([i])).digest(128))) for i in range(3)] | |
| e_hat = [cls.ntt(cls.sample_cbd2(hashlib.shake_256(sigma + bytes([i + 3])).digest(128))) for i in range(3)] | |
| t_hat = [] | |
| for i in range(3): | |
| ti = cls.basemul_acc(a_hat[i], s_hat) | |
| t_hat.append([(a + b) % 3329 for a, b in zip(ti, e_hat[i])]) | |
| ek_pke = b"".join(cls.encode_12(p) for p in t_hat) + rho | |
| dk_pke = b"".join(cls.encode_12(p) for p in s_hat) | |
| h = hashlib.sha3_256(ek_pke).digest() | |
| return ek_pke, dk_pke + ek_pke + h + z | |
| def _pke_encrypt_expanded( | |
| cls, | |
| t_hat: List[List[int]], | |
| a_hat_t: List[List[List[int]]], | |
| m: bytes, | |
| r: bytes, | |
| ) -> bytes: | |
| r_hat = [cls.ntt(cls.sample_cbd2(hashlib.shake_256(r + bytes([i])).digest(128))) for i in range(3)] | |
| e1 = [cls.sample_cbd2(hashlib.shake_256(r + bytes([i + 3])).digest(128)) for i in range(3)] | |
| e2 = cls.sample_cbd2(hashlib.shake_256(r + bytes([6])).digest(128)) | |
| u = [] | |
| for i in range(3): | |
| ui = cls.intt(cls.basemul_acc(a_hat_t[i], r_hat)) | |
| u.append([(a + b) % 3329 for a, b in zip(ui, e1[i])]) | |
| mu = cls.encode_1(m) | |
| v_ntt = cls.basemul_acc(t_hat, r_hat) | |
| v_poly = cls.intt(v_ntt) | |
| v = [(a + b + c) % 3329 for a, b, c in zip(v_poly, e2, mu)] | |
| return b"".join(cls.compress_encode_10(p) for p in u) + cls.compress_encode_4(v) | |
| def encaps(cls, ek: bytes, seed_m: Optional[bytes] = None) -> Tuple[bytes, bytes]: | |
| if len(ek) != cls.EK_BYTES: | |
| raise ValueError(f"ML-KEM-768 ek must be {cls.EK_BYTES} bytes") | |
| if ek[:1152] == b"\x00" * 1152 or ek[1152:] == b"\x00" * 32: | |
| raise ValueError("ML-KEM-768 degenerate all-zero encapsulation key rejected") | |
| t_hat, a_hat_t, ek_hash = cls._expand_ek(ek) | |
| if b"".join(cls.encode_12(p) for p in t_hat) != ek[:1152]: | |
| raise ValueError("ML-KEM-768 non-canonical encapsulation key (coefficients >= q)") | |
| m = seed_m if seed_m is not None else os.urandom(32) | |
| kr = hashlib.sha3_512(m + ek_hash).digest() | |
| k_shared, r = kr[:32], kr[32:] | |
| c = cls._pke_encrypt_expanded(t_hat, a_hat_t, m, r) | |
| return c, k_shared | |
| def decaps(cls, dk: bytes, c: bytes) -> bytes: | |
| if len(dk) != cls.DK_BYTES or len(c) != cls.CT_BYTES: | |
| raise ValueError("ML-KEM-768 invalid dk or ciphertext length") | |
| s_hat, _ek, h, z, t_hat, a_hat_t = cls._expand_dk(dk) | |
| u = [cls.decode_decompress_10(c[320 * i : 320 * (i + 1)]) for i in range(3)] | |
| v = cls.decode_decompress_4(c[960:1088]) | |
| u_hat = [cls.ntt(p) for p in u] | |
| s_u = cls.intt(cls.basemul_acc(s_hat, u_hat)) | |
| w = [(vi - sui) % 3329 for vi, sui in zip(v, s_u)] | |
| m_prime = cls.decode_1(w) | |
| kr_prime = hashlib.sha3_512(m_prime + h).digest() | |
| k_prime, r_prime = kr_prime[:32], kr_prime[32:] | |
| k_bar = hashlib.shake_256(z + c).digest(32) | |
| c_prime = cls._pke_encrypt_expanded(t_hat, a_hat_t, m_prime, r_prime) | |
| if hmac.compare_digest(c, c_prime): | |
| return k_prime | |
| return k_bar | |
| MLKEM768._init_tables() | |
| class WOTSPlusSHAKE256: | |
| """Post-Quantum Winternitz Hash-Based Signature over SHAKE-256 (RFC 8391 / NIST FIPS 205 building block).""" | |
| N = 32 | |
| W = 16 | |
| CHAINS = 67 | |
| PRIV_BYTES = 64 | |
| PUB_BYTES = 32 | |
| SIG_BYTES = 2176 | |
| def _chain(x: bytes, start: int, steps: int, pub_seed: bytes, chain_idx: int) -> bytes: | |
| cur = x | |
| prefix = pub_seed + bytes([chain_idx]) | |
| for step in range(start, start + steps): | |
| cur = hashlib.shake_256(prefix + bytes([step]) + cur).digest(32) | |
| return cur | |
| def _msg_to_base_w(cls, msg_digest_32: bytes) -> List[int]: | |
| digits: List[int] = [] | |
| for b in msg_digest_32: | |
| digits.append((b >> 4) & 0x0F) | |
| digits.append(b & 0x0F) | |
| csum = sum(15 - d for d in digits) | |
| digits.append((csum >> 8) & 0x0F) | |
| digits.append((csum >> 4) & 0x0F) | |
| digits.append(csum & 0x0F) | |
| return digits | |
| def generate_keypair(cls, seed: Optional[bytes] = None) -> Tuple[bytes, bytes]: | |
| raw = seed if seed is not None else os.urandom(64) | |
| return cls._generate_keypair_from_seed(raw) | |
| def _generate_keypair_from_seed(cls, raw: bytes) -> Tuple[bytes, bytes]: | |
| if len(raw) == 32: | |
| raw = hashlib.shake_256(b"armiger-wots-keygen-v1:" + raw).digest(64) | |
| elif len(raw) != 64: | |
| raise ValueError("WOTS+-SHAKE256 seed must be 32 or 64 bytes") | |
| sk_seed, pub_seed = raw[:32], raw[32:] | |
| ends: List[bytes] = [] | |
| for i in range(cls.CHAINS): | |
| sk_i = hashlib.shake_256(sk_seed + bytes([i])).digest(32) | |
| ends.append(cls._chain(sk_i, 0, 15, pub_seed, i)) | |
| pk = hashlib.shake_256(pub_seed + b"".join(ends)).digest(32) | |
| return raw, pk | |
| def derive_keypair_from_ed25519_priv(cls, ed25519_priv_bytes: bytes, seq: int = 0) -> Tuple[bytes, bytes]: | |
| ctx = b"armiger-hybrid-ed25519-wots-v1:" + ed25519_priv_bytes + (f":seq:{int(seq)}".encode("utf-8") if seq else b"") | |
| seed64 = hashlib.shake_256(ctx).digest(64) | |
| return cls.generate_keypair(seed64) | |
| def sign(cls, priv_seed: bytes, message: bytes) -> bytes: | |
| if len(priv_seed) == 32: | |
| priv_seed, pk = cls.derive_keypair_from_ed25519_priv(priv_seed) | |
| else: | |
| _, pk = cls.generate_keypair(priv_seed) | |
| sk_seed, pub_seed = priv_seed[:32], priv_seed[32:] | |
| digest = hashlib.shake_256(b"armiger-wots-msg-v2:" + pub_seed + pk + message).digest(32) | |
| digits = cls._msg_to_base_w(digest) | |
| sig_parts = [pub_seed] | |
| for i, d in enumerate(digits): | |
| sk_i = hashlib.shake_256(sk_seed + bytes([i])).digest(32) | |
| sig_parts.append(cls._chain(sk_i, 0, d, pub_seed, i)) | |
| return b"".join(sig_parts) | |
| def verify(cls, pub_key: bytes, message: bytes, signature: bytes) -> bool: | |
| if len(pub_key) != cls.PUB_BYTES or len(signature) != cls.SIG_BYTES: | |
| return False | |
| pub_seed = signature[:32] | |
| digest = hashlib.shake_256(b"armiger-wots-msg-v2:" + pub_seed + pub_key + message).digest(32) | |
| digits = cls._msg_to_base_w(digest) | |
| ends: List[bytes] = [] | |
| for i, d in enumerate(digits): | |
| sig_i = signature[32 * (i + 1) : 32 * (i + 2)] | |
| ends.append(cls._chain(sig_i, d, 15 - d, pub_seed, i)) | |
| computed_pk = hashlib.shake_256(pub_seed + b"".join(ends)).digest(32) | |
| return hmac.compare_digest(computed_pk, pub_key) | |
| # --------------------------------------------------------------------------- | |
| # Hybrid Ed25519 + WOTS+-SHAKE256 Blinded SHA-256 Tombstone Purge Engine | |
| # --------------------------------------------------------------------------- | |
| def blind_infohash(infohash: str) -> str: | |
| """Compute deterministic Post-Quantum blinded SHA-256 hash for an infohash. | |
| Formula: SHA-256("armiger-tombstone-v1:" || lowercase_hex_infohash) | |
| Preserves anonymity while allowing deterministic matching on known bad hashes (128-bit Grover PQ preimage resistance). | |
| """ | |
| clean = str(infohash).strip().lower().encode("utf-8", errors="replace") | |
| return hashlib.sha256(b"armiger-tombstone-v1:" + clean).hexdigest() | |
| class SignedTombstoneList: | |
| """Hybrid Post-Quantum (Ed25519 + WOTS+-SHAKE256) signed revocation manifest.""" | |
| seq: int | |
| timestamp: int | |
| tombstones: list[str] | |
| reasons: dict[str, str] | |
| signature_hex: str | |
| pq_pubkey_hex: str = "" | |
| pq_signature_hex: str = "" | |
| def canonical_payload(self) -> bytes: | |
| """Deterministic, sorted JSON bytes for Hybrid Ed25519 + WOTS+-SHAKE256 signature verification.""" | |
| body = { | |
| "reasons": {k: self.reasons.get(k, "CSAM_ABUSE") for k in sorted(self.tombstones)}, | |
| "seq": int(self.seq), | |
| "timestamp": int(self.timestamp), | |
| "tombstones": sorted([t.lower() for t in self.tombstones]), | |
| } | |
| return json.dumps(body, sort_keys=True, separators=(",", ":")).encode("utf-8") | |
| class TombstoneStore: | |
| """Maintains active blinded tombstone revocation set and verifies Hybrid Ed25519 + WOTS+-SHAKE256 updates.""" | |
| def __init__( | |
| self, | |
| trusted_root_pubkeys_hex: Sequence[str] | None = None, | |
| initial_seq: int = 0, | |
| trusted_root_pq_pubkeys_hex: Sequence[str] | None = None, | |
| ) -> None: | |
| self.trusted_pubkeys = [ | |
| bytes.fromhex(k.strip().lower()) | |
| for k in (trusted_root_pubkeys_hex or []) | |
| if len(k.strip()) == 64 | |
| ] | |
| self.trusted_pq_pubkeys = [ | |
| bytes.fromhex(k.strip().lower()) | |
| for k in (trusted_root_pq_pubkeys_hex or []) | |
| if len(k.strip()) == 64 | |
| ] | |
| self.current_seq = int(initial_seq) | |
| self.blinded_hashes: set[str] = set() | |
| self.reasons: dict[str, str] = {} | |
| def add_trusted_pubkey(self, pubkey_hex: str, pq_pubkey_hex: str = "") -> None: | |
| raw = bytes.fromhex(pubkey_hex.strip().lower()) | |
| if len(raw) == 32 and raw not in self.trusted_pubkeys: | |
| self.trusted_pubkeys.append(raw) | |
| if pq_pubkey_hex: | |
| raw_pq = bytes.fromhex(pq_pubkey_hex.strip().lower()) | |
| if len(raw_pq) == 32 and raw_pq not in self.trusted_pq_pubkeys: | |
| self.trusted_pq_pubkeys.append(raw_pq) | |
| def is_tombstoned(self, infohash: str) -> bool: | |
| """Check whether an infohash matches any active blinded tombstone.""" | |
| blinded = blind_infohash(infohash) | |
| return blinded in self.blinded_hashes | |
| def apply_signed_tombstones(self, manifest: SignedTombstoneList) -> tuple[bool, str, int]: | |
| """Verify Hybrid Ed25519 + WOTS+-SHAKE256 signature, monotonic sequence, and ingest new tombstones. | |
| Returns (success: bool, message: str, new_tombstones_count: int). | |
| """ | |
| if manifest.seq <= self.current_seq: | |
| return False, f"stale_or_replayed_tombstone_seq:{manifest.seq}<={self.current_seq}", 0 | |
| payload = manifest.canonical_payload() | |
| try: | |
| sig_bytes = bytes.fromhex(manifest.signature_hex) | |
| except ValueError: | |
| return False, "ed25519_signature_verification_failed", 0 | |
| # 1. Classical Ed25519 signature verification against trusted roots | |
| verified = False | |
| if not self.trusted_pubkeys: | |
| return False, "no_trusted_root_keys_configured", 0 | |
| for pub_bytes in self.trusted_pubkeys: | |
| try: | |
| vk = Ed25519PublicKey.from_public_bytes(pub_bytes) | |
| vk.verify(sig_bytes, payload) | |
| verified = True | |
| break | |
| except (InvalidSignature, ValueError): | |
| continue | |
| if not verified: | |
| return False, "ed25519_signature_verification_failed", 0 | |
| # 2. Post-Quantum WOTS+-SHAKE256 signature verification | |
| if manifest.pq_signature_hex or manifest.pq_pubkey_hex or self.trusted_pq_pubkeys: | |
| try: | |
| pq_sig_bytes = bytes.fromhex(manifest.pq_signature_hex) | |
| except ValueError: | |
| return False, "pq_wots_signature_verification_failed", 0 | |
| candidate_pq_pubs: List[bytes] = [] | |
| if self.trusted_pq_pubkeys: | |
| candidate_pq_pubs.extend(self.trusted_pq_pubkeys) | |
| elif manifest.pq_pubkey_hex: | |
| try: | |
| candidate_pq_pubs.append(bytes.fromhex(manifest.pq_pubkey_hex.strip().lower())) | |
| except ValueError: | |
| return False, "pq_wots_signature_verification_failed", 0 | |
| pq_verified = any( | |
| WOTSPlusSHAKE256.verify(pq_pub, payload, pq_sig_bytes) | |
| for pq_pub in candidate_pq_pubs | |
| ) | |
| if not pq_verified: | |
| return False, "pq_wots_signature_verification_failed", 0 | |
| # Monotonic update | |
| self.current_seq = manifest.seq | |
| new_count = 0 | |
| for thash in manifest.tombstones: | |
| clean_hash = thash.lower() | |
| if clean_hash not in self.blinded_hashes: | |
| self.blinded_hashes.add(clean_hash) | |
| self.reasons[clean_hash] = manifest.reasons.get(clean_hash, "CSAM_ABUSE") | |
| new_count += 1 | |
| return True, "verified_and_applied", new_count | |
| def generate_ed25519_keypair() -> tuple[bytes, bytes]: | |
| """Generate a fresh Ed25519 private key (32 bytes) and public key (32 bytes).""" | |
| priv = Ed25519PrivateKey.generate() | |
| pub = priv.public_key() | |
| from cryptography.hazmat.primitives.serialization import Encoding, PublicFormat, PrivateFormat, NoEncryption | |
| priv_bytes = priv.private_bytes(Encoding.Raw, PrivateFormat.Raw, NoEncryption()) | |
| pub_bytes = pub.public_bytes(Encoding.Raw, PublicFormat.Raw) | |
| return priv_bytes, pub_bytes | |
| def sign_tombstone_list( | |
| seq: int, | |
| infohashes_or_blinded: Sequence[str], | |
| private_key_bytes: bytes, | |
| reasons: dict[str, str] | None = None, | |
| are_blinded: bool = False, | |
| timestamp: int | None = None, | |
| ) -> SignedTombstoneList: | |
| """Create and dual-sign a SignedTombstoneList manifest with Ed25519 + Post-Quantum WOTS+-SHAKE256.""" | |
| ts = timestamp if timestamp is not None else int(time.time()) | |
| blinded_list = [ | |
| h.lower() if are_blinded else blind_infohash(h) | |
| for h in infohashes_or_blinded | |
| ] | |
| reasons_map = {} | |
| for h in blinded_list: | |
| reasons_map[h] = (reasons or {}).get(h, "CSAM_ABUSE") | |
| dummy = SignedTombstoneList( | |
| seq=seq, | |
| timestamp=ts, | |
| tombstones=blinded_list, | |
| reasons=reasons_map, | |
| signature_hex="", | |
| ) | |
| payload = dummy.canonical_payload() | |
| sk = Ed25519PrivateKey.from_private_bytes(private_key_bytes) | |
| sig = sk.sign(payload) | |
| pq_priv, pq_pub = WOTSPlusSHAKE256.derive_keypair_from_ed25519_priv(private_key_bytes, seq=seq) | |
| pq_sig = WOTSPlusSHAKE256.sign(pq_priv, payload) | |
| return SignedTombstoneList( | |
| seq=seq, | |
| timestamp=ts, | |
| tombstones=blinded_list, | |
| reasons=reasons_map, | |
| signature_hex=sig.hex(), | |
| pq_pubkey_hex=pq_pub.hex(), | |
| pq_signature_hex=pq_sig.hex(), | |
| ) | |