/ custom_types / NIP_35.py
NIP_35.py
1 import re 2 from typing import Literal, Optional 3 4 from pydantic import BaseModel, Field, field_validator, model_validator 5 6 # --- Helpers (strict, wire-level) --- 7 8 HEX_64 = re.compile(r"^[0-9a-f]{64}$", re.IGNORECASE) 9 SIG_128 = re.compile(r"^[0-9a-f]{128}$", re.IGNORECASE) 10 11 # NIP-35: "x" is V1 BitTorrent Info Hash as in magnet:?xt=urn:btih:HASH 12 # btih can be 40-hex (20 bytes) or base32 (32 chars). We'll accept both. 13 BTIH_V1 = re.compile(r"^(?:[0-9a-f]{40}|[A-Z2-7]{32})$", re.IGNORECASE) 14 15 # tags[0] must be one of these for kind 2003, per the example/spec 16 ALLOWED_TAG_KINDS = {"title", "x", "file", "tracker", "i", "t", "size"} 17 18 # NIP-35 "i" tag prefixes allowed 19 ALLOWED_I_PREFIXES = {"tcat", "newznab", "tmdb", "ttvdb", "imdb", "mal", "anilist"} 20 21 22 class Nip35Kind2003Event(BaseModel): 23 """ 24 Validates a NIP-35 Torrent event (kind=2003) in wire format: 25 { 26 "id": "...", 27 "pubkey": "...", 28 "created_at": 123, 29 "kind": 2003, 30 "tags": [ ["title", "..."], ["x", "..."], ["file", "...", "..."], ... ], 31 "content": "...", 32 "sig": "..." 33 } 34 35 Enforces: 36 - kind == 2003 37 - tags are list[list[str]] with known tag types and correct arity 38 - required tags exist: >=1 title, >=1 x, >=1 file 39 - "x" matches btih v1 (hex40 or base32-32) 40 - "file" size is a decimal string 41 - "i" prefix is one of the NIP-35 prefixes (and supports second-level prefix) 42 :contentReference[oaicite:1]{index=1} 43 """ 44 45 id: str 46 pubkey: str 47 created_at: int = Field(ge=0) 48 kind: Literal[2003] 49 tags: list[list[str]] 50 content: str 51 sig: Optional[str] = None 52 53 # --- Basic Nostr-ish sanity checks (optional but usually desirable) --- 54 @field_validator("id") 55 @classmethod 56 def _id_hex64(cls, v: str) -> str: 57 if not HEX_64.match(v): 58 raise ValueError("id must be 32-byte hex (64 hex chars)") 59 return v 60 61 @field_validator("pubkey") 62 @classmethod 63 def _pubkey_hex64(cls, v: str) -> str: 64 if not HEX_64.match(v): 65 raise ValueError("pubkey must be 32-byte hex (64 hex chars)") 66 return v 67 68 @field_validator("sig") 69 @classmethod 70 def _sig_hex128(cls, v: Optional[str]) -> Optional[str]: 71 if v is not None and not SIG_128.match(v): 72 raise ValueError("sig must be 64-byte hex (128 hex chars)") 73 return v 74 75 @field_validator("tags") 76 @classmethod 77 def _tags_are_lists_of_str(cls, tags: list[list[str]]) -> list[list[str]]: 78 # Ensure each tag is a non-empty list of strings 79 for i, tag in enumerate(tags): 80 if not isinstance(tag, list) or len(tag) == 0: 81 raise ValueError(f"tag[{i}] must be a non-empty list") 82 if not all(isinstance(x, str) for x in tag): 83 raise ValueError(f"tag[{i}] must contain only strings") 84 return tags 85 86 @model_validator(mode="after") 87 def _validate_nip35_kind2003(self) -> "Nip35Kind2003Event": 88 seen_title = 0 89 seen_x = 0 90 seen_file = 0 91 92 for idx, tag in enumerate(self.tags): 93 tag_kind = tag[0] 94 95 if tag_kind not in ALLOWED_TAG_KINDS: 96 raise ValueError( 97 f"tag[{idx}][0]={tag_kind!r} not allowed for NIP-35 kind=2003" 98 ) 99 100 # ["title", "<torrent-title>"] 101 if tag_kind == "title": 102 if len(tag) != 2 or not tag[1]: 103 raise ValueError(f"tag[{idx}] must be ['title', <non-empty str>]") 104 seen_title += 1 105 106 # ["x", "<bittorrent-info-hash>"] 107 elif tag_kind == "x": 108 if len(tag) != 2: 109 raise ValueError(f"tag[{idx}] must be ['x', <hash>]") 110 if not BTIH_V1.match(tag[1]): 111 raise ValueError( 112 f"tag[{idx}] 'x' must be btih v1 (40-hex or base32-32 chars)" 113 ) 114 seen_x += 1 115 116 # ["file", "<file-name>", "<file-size-in-bytes>"] 117 elif tag_kind == "file": 118 if len(tag) != 3: 119 raise ValueError( 120 f"tag[{idx}] must be ['file', <path>, <size-bytes>]" 121 ) 122 if not tag[1]: 123 raise ValueError(f"tag[{idx}] file path must be non-empty") 124 if not tag[2].isdigit(): 125 raise ValueError( 126 f"tag[{idx}] file size must be a decimal string (bytes)" 127 ) 128 seen_file += 1 129 130 # ["tracker", "<url>"] (optional, repeatable) 131 elif tag_kind == "tracker": 132 if len(tag) != 2 or not tag[1]: 133 raise ValueError( 134 f"tag[{idx}] must be ['tracker', <non-empty url str>]" 135 ) 136 # spec examples use udp:// and http://; keep it permissive 137 if "://" not in tag[1]: 138 raise ValueError( 139 f"tag[{idx}] tracker should look like a URL (contain '://')" 140 ) 141 142 # ["i", "<prefix>:..."] (optional, repeatable) 143 elif tag_kind == "i": 144 if len(tag) != 2 or not tag[1]: 145 raise ValueError( 146 f"tag[{idx}] must be ['i', '<prefix>:...']" 147 ) 148 # allow second-level prefix (tmdb:movie:..., mal:anime:..., etc.) 149 prefix = tag[1].split(":", 1)[0] 150 if prefix not in ALLOWED_I_PREFIXES: 151 raise ValueError( 152 f"tag[{idx}] i-prefix {prefix!r} not in {sorted(ALLOWED_I_PREFIXES)}" 153 ) 154 155 # ["t", "<category>"] (optional, repeatable) 156 elif tag_kind == "t": 157 if len(tag) != 2 or not tag[1]: 158 raise ValueError(f"tag[{idx}] must be ['t', <non-empty str>]") 159 160 # Required by the NIP-35 kind=2003 definition/example: title, x, at least one file 161 # (the NIP defines the tags needed to construct a magnet link and search content). :contentReference[oaicite:2]{index=2} 162 if seen_title < 1: 163 raise ValueError("missing required tag: ['title', ...]") 164 if seen_x < 1: 165 raise ValueError("missing required tag: ['x', ...]") 166 if seen_file < 1: 167 raise ValueError("missing required tag: ['file', ...] (at least one)") 168 169 return self