Coverage for lode / reader / security.py: 94%
123 statements
« prev ^ index » next coverage.py v7.13.0, created at 2026-10-05 16:02 +0000
1"""Security validation for incoming semantic artefacts."""
2import os
3import socket
4import ipaddress
5from urllib.parse import urlparse
6from lode.exceptions import ArtefactValidationError
9def _env_int(name: str, default: int) -> int:
10 try:
11 return int(os.getenv(name, str(default)))
12 except (TypeError, ValueError):
13 return default
16def _user_agent() -> str:
17 """Identify LODE to remote hosts.
19 The requests default ("python-requests/x.y") is rejected outright by the
20 bot filters in front of some ontology registries, which reaches the user
21 as a bogus 502 on an artefact that is actually available.
22 """
23 try:
24 from importlib.metadata import version
25 v = version("lode")
26 except Exception:
27 v = "0"
28 return os.getenv(
29 "LODE_USER_AGENT",
30 f"LODE/{v} (+https://github.com/opencitations/lode)",
31 )
34USER_AGENT = _user_agent()
35MAX_BYTES = _env_int("LODE_MAX_UPLOAD_MB", 10) * 1024 * 1024 # 10 MB default
36MAX_REDIRECTS = _env_int("LODE_MAX_REDIRECTS", 6) # hops followed when fetching a URL
37MAX_ENTITY_DECLARATIONS = _env_int("LODE_MAX_XML_ENTITIES", 100)
38MAX_DTD_SUBSET = 50 * 1024 # bytes; a legitimate DOCTYPE internal subset is tiny
39ALLOWED_EXTENSIONS = {".rdf", ".owl", ".ttl", ".n3", ".nt", ".jsonld", ".xml"}
40ALLOWED_SCHEMES = {"http", "https"}
42# Control characters that are legitimate in a text file.
43_ALLOWED_CONTROL = {"\t", "\n", "\r", "\f", "\v"}
45# Magic-byte signatures of common binary/archive/executable formats.
46_BINARY_SIGNATURES: tuple[tuple[bytes, str], ...] = (
47 (b"%PDF", "PDF"),
48 (b"PK\x03\x04", "ZIP/Office"),
49 (b"PK\x05\x06", "ZIP"),
50 (b"PK\x07\x08", "ZIP"),
51 (b"\x1f\x8b", "gzip"),
52 (b"BZh", "bzip2"),
53 (b"\xfd7zXZ\x00", "xz"),
54 (b"7z\xbc\xaf\x27\x1c", "7-Zip"),
55 (b"Rar!\x1a\x07", "RAR"),
56 (b"\x89PNG\r\n\x1a\n", "PNG"),
57 (b"\xff\xd8\xff", "JPEG"),
58 (b"GIF87a", "GIF"),
59 (b"GIF89a", "GIF"),
60 (b"\x7fELF", "ELF executable"),
61 (b"MZ", "Windows executable"),
62 (b"\xca\xfe\xba\xbe", "Mach-O/Java class"),
63 (b"\xcf\xfa\xed\xfe", "Mach-O"),
64 (b"\xce\xfa\xed\xfe", "Mach-O"),
65 (b"\xfe\xed\xfa\xce", "Mach-O"),
66 (b"\xfe\xed\xfa\xcf", "Mach-O"),
67 (b"%!PS", "PostScript"),
68 (b"SQLite format 3\x00", "SQLite database"),
69)
71def _has_entity_reference(text: str) -> bool:
72 """True if text contains a general entity reference like '&name;' (ignoring
73 numeric character references such as 'A'). Pure string scan with a
74 bounded name length, so it is linear and ReDoS-free."""
75 j = text.find("&")
76 while j != -1:
77 k = j + 1
78 if k < len(text) and text[k] != "#" and (text[k].isalpha() or text[k] == "_"):
79 m = k + 1
80 end = min(len(text), m + 64) # entity names are short; bound the scan
81 while m < end and (text[m].isalnum() or text[m] in "_.-"):
82 m += 1
83 if m < len(text) and text[m] == ";":
84 return True
85 j = text.find("&", j + 1)
86 return False
89def check_size(num_bytes: int) -> None:
90 if num_bytes > MAX_BYTES:
91 raise ArtefactValidationError("File too large", context={"bytes": num_bytes, "max": MAX_BYTES})
93def check_extension(name: str) -> None:
94 if not name:
95 raise ArtefactValidationError("Missing filename", context={"name": name})
96 ext = os.path.splitext(urlparse(name).path)[1].lower()
97 if ext not in ALLOWED_EXTENSIONS:
98 raise ArtefactValidationError("Extension not allowed", context={"ext": ext})
100def check_ip_safe(ip_str: str) -> None:
101 """Reject an IP that points at an internal/non-routable range (SSRF).
103 Reused both before connecting (each resolved IP) and after connecting (the
104 real peer IP), so a DNS rebinding between the two cannot reach an internal
105 host.
106 """
107 ip = ipaddress.ip_address(ip_str)
108 # IPv4-mapped IPv6 (e.g. ::ffff:127.0.0.1) would bypass the v4 checks below.
109 if getattr(ip, "ipv4_mapped", None):
110 ip = ip.ipv4_mapped
111 if (ip.is_private or ip.is_loopback or ip.is_link_local
112 or ip.is_reserved or ip.is_multicast or ip.is_unspecified):
113 raise ArtefactValidationError("Blocked address", context={"ip": str(ip)})
116def check_url_safe(url: str) -> None:
117 """Block non-http schemes and SSRF toward private/internal hosts."""
118 parsed = urlparse(url)
119 if parsed.scheme not in ALLOWED_SCHEMES:
120 raise ArtefactValidationError("Scheme not allowed", context={"scheme": parsed.scheme})
121 host = parsed.hostname
122 if not host:
123 raise ArtefactValidationError("Missing host", context={"url": url})
124 try:
125 infos = socket.getaddrinfo(host, None)
126 except socket.gaierror:
127 raise ArtefactValidationError("Cannot resolve host", context={"host": host})
128 for info in infos:
129 check_ip_safe(info[4][0])
131def check_is_text(data: bytes) -> None:
132 """RDF serializations are text. Reject binary blobs."""
133 if not data:
134 raise ArtefactValidationError("Empty artefact")
135 # 1) known binary/archive/executable signature -> reject up front
136 head = data[:64]
137 for signature, label in _BINARY_SIGNATURES:
138 if head.startswith(signature):
139 raise ArtefactValidationError(
140 "Not a text/ASCII artefact (binary content)", context={"detected": label}
141 )
142 # 2) NUL byte is a strong binary signal
143 if b"\x00" in data:
144 raise ArtefactValidationError("Not a text/ASCII artefact (binary content)")
145 # 3) must be valid UTF-8 (handle a leading BOM)
146 try:
147 text = data.decode("utf-8-sig")
148 except UnicodeDecodeError:
149 raise ArtefactValidationError("Not a text/ASCII artefact")
150 # 4) too many non-printable control chars -> not a real RDF serialization
151 sample = text[:8192]
152 if sample:
153 control = sum(1 for ch in sample if ord(ch) < 32 and ch not in _ALLOWED_CONTROL)
154 if control / len(sample) > 0.05:
155 raise ArtefactValidationError("Not a text/ASCII artefact (control characters)")
157def check_safe_xml(text: str) -> None:
158 """Block dangerous XML documents while keeping simple internal entities.
160 Legitimate RDF/XML (e.g. Protege exports) often uses a DOCTYPE with internal
161 entities to abbreviate namespaces, so we cannot forbid every DTD. We forbid
162 only the exploitable patterns:
164 * EXTERNAL DTD/entities (SYSTEM/PUBLIC) -> XXE, file read, SSRF
165 * PARAMETER entities (<!ENTITY % ...>) -> advanced XXE
166 * entities referencing other entities -> billion laughs
167 * too many entity declarations -> abuse / expansion
168 """
169 lowered = text.lower()
170 idx = lowered.find("<!doctype")
171 if idx == -1:
172 return # no DOCTYPE => no DTD surface
174 end = text.find(">", idx)
175 bracket = text.find("[", idx)
176 if bracket != -1:
177 close = text.find("]", bracket)
178 subset = text[bracket: close + 1 if close != -1 else len(text)]
179 else:
180 subset = text[idx: end + 1 if end != -1 else len(text)]
182 # A real DOCTYPE internal subset is tiny. An oversized one is both suspicious
183 # and a ReDoS vector for the scans below, so reject it outright.
184 if len(subset) > MAX_DTD_SUBSET:
185 raise ArtefactValidationError("DOCTYPE internal subset too large")
187 subset_lower = subset.lower()
188 if "system" in subset_lower or "public" in subset_lower:
189 raise ArtefactValidationError("XML with external DTD/entity not allowed (XXE risk)")
191 # Analyse the declarations with plain string ops only (no regex on the
192 # user-controlled subset -> no ReDoS): count them, flag parameter entities,
193 # and reject any nested entity reference (billion laughs).
194 if subset_lower.count("<!entity") > MAX_ENTITY_DECLARATIONS:
195 raise ArtefactValidationError("Too many XML entity declarations (entity-expansion)")
196 i = subset_lower.find("<!entity")
197 while i != -1:
198 after = subset_lower[i + len("<!entity"):].lstrip()
199 if after.startswith("%"): # parameter entity: <!ENTITY % name ...>
200 raise ArtefactValidationError("XML parameter entity not allowed (XXE risk)")
201 i = subset_lower.find("<!entity", i + len("<!entity"))
202 if _has_entity_reference(subset): # an entity value references another entity
203 raise ArtefactValidationError("Nested XML entities not allowed (billion laughs)")
205async def read_upload_capped(upload, max_bytes: int = MAX_BYTES) -> bytes:
206 """Read an UploadFile in chunks, aborting as soon as it exceeds max_bytes.
208 Avoids loading an arbitrarily large upload fully into memory before the size
209 check (which `await file.read()` would do).
210 """
211 chunks: list[bytes] = []
212 total = 0
213 while True:
214 chunk = await upload.read(64 * 1024)
215 if not chunk:
216 break
217 total += len(chunk)
218 if total > max_bytes:
219 raise ArtefactValidationError("File too large", context={"max": max_bytes})
220 chunks.append(chunk)
221 return b"".join(chunks)