Coverage for lode / reader / loader.py: 93%
124 statements
« prev ^ index » next coverage.py v7.13.0, created at 2026-10-05 16:02 +0000
1"""
2Reader - Caricamento e validazione file RDF
3"""
4import requests
5import os
6from rdflib import Graph
7from typing import Dict, Optional
8from urllib.parse import urlparse, urljoin
10import lode.reader.modules as modules
11from lode.reader import security
12from lode.exceptions import (ArtefactLoadError, ArtefactNotFoundError,
13 ArtefactUnavailableError, ArtefactValidationError)
14from lode.helpers.rdf import PARSE_FORMATS, parse_any
15from lode.reader.provenance_graph import ProvenanceGraph
17class Loader:
18 """Gestisce il caricamento di file RDF"""
20 def __init__(self, file_path: Optional[str] = None, imported=None, closure=None, partial_import=None):
22 self.graph = Graph()
23 self.ontology_iri = None
24 self._imported = imported
25 self._closure = closure
26 self.imported_uris = set() # needed to keep track of original model to document
27 self._partial_import = partial_import
29 if file_path:
30 self.load(file_path)
32 # ----------------------------------------------------------
33 # MAIN LOAD METHOD
34 # ----------------------------------------------------------
35 def load(self, source: str) -> None:
36 """Loads RDF from local file or from URL with content negotiation"""
38 if self._is_url(source):
39 self._load_from_url_with_content_negotiation(source)
41 # ADD BY VALE TO FIX WINDOWS LOCAL BUG - COMMENTED FOR PRODUCTION
42 elif os.path.isfile(source):
43 self._load_from_local_file(source)
44 else:
45 # A value carrying a URL scheme that is not http(s) (file:, ftp:, ...)
46 # must not be silently treated as a local path. Bare local paths
47 # (no scheme) are still allowed here for the CLI.
48 scheme = urlparse(source).scheme
49 if scheme:
50 raise ArtefactValidationError(
51 "URL scheme not allowed; use http(s)://host",
52 context={"scheme": scheme},
53 )
54 self._load_from_local_file(source)
56 if len(self.graph) == 0:
57 raise ArtefactLoadError(
58 "Parsed graph is empty (wrong URL or not an RDF resource)",
59 context={"source": source}
60 )
62 if self._imported or self._closure or self._partial_import:
63 raw, self.graph = self.graph, ProvenanceGraph()
64 self.ontology_iri = self.graph.add_document(raw, source)
65 self._apply_modules()
67 # ----------------------------------------------------------
68 # MODULES MAIN HANDLER
69 # ----------------------------------------------------------
71 def _apply_modules(self) -> None:
72 if self._partial_import:
73 self.graph, self.imported_uris = modules.apply_partial_import(self.graph)
74 elif self._imported:
75 self.graph, self.imported_uris = modules.apply_imported(self.graph)
76 elif self._closure:
77 self.graph, self.imported_uris = modules.apply_closure(self.graph)
79 ## ----------------------------------------------------------
80 # CONTENT NEGOTIATION FOR URLS
81 # ----------------------------------------------------------
82 def _load_from_url_with_content_negotiation(self, url: str) -> None:
83 """RDF graph loading with content-negotiation for semantic artefact loaded from URL"""
85 security.check_url_safe(url) # SECURITY: url validation (anti-SSRF), before downlaod
87 headers = {
88 "Accept": (
89 "text/turtle, application/rdf+xml, application/ld+json, "
90 "application/n-triples, application/n-quads, */*;q=0.1"
91 ),
92 # Sent on every redirect hop too, since this dict is reused there.
93 "User-Agent": security.USER_AGENT,
94 }
96 response = None
97 try:
98 # SECURITY: fetch with per-hop URL validation (anti-SSRF); manual redirects, no auto-follow into unchecked hosts
99 response = self._fetch_following_redirects(url, headers)
101 # Error: Cannot Load RDF. A 5xx says the host is broken right now,
102 # not that the artefact is missing: telling the two apart is the
103 # difference between "check your URL" and "try again later".
104 if response.status_code != 200:
105 if response.status_code >= 500:
106 raise ArtefactUnavailableError(
107 "The remote server is not responding, please try again later",
108 context={"url": url, "http_status": response.status_code}
109 )
110 raise ArtefactNotFoundError(
111 "Cannot load provided Semantic Artefact",
112 context={"url": url, "http_status": response.status_code}
113 )
115 # SECURITY: reject by declared Content-Length before downloading the body
116 declared = response.headers.get("Content-Length")
117 if declared and declared.isdigit():
118 security.check_size(int(declared))
120 # SECURITY: stop if file > 10MB even without Content-Length
121 raw = b""
122 for chunk in response.iter_content(8192): # 8kb at the time in iteration
123 raw += chunk
124 if len(raw) > security.MAX_BYTES:
125 security.check_size(len(raw))
127 # SECURITY: checks the file is text not binary
128 security.check_is_text(raw)
130 # Proceed with encoding
131 content = raw.decode("utf-8")
133 # SECURITY: block XXE / entity-expansion on remote artefacts and owl:imports
134 security.check_safe_xml(content)
136 content_type = response.headers.get("Content-Type", "").lower()
138 # Format guessed from HTTP Content-Type (content negotiation handler)
139 graph = parse_any(content, preferred=self._guess_format_from_content_type(content_type))
140 if graph is None:
141 raise ArtefactLoadError("Could not parse RDF after content negotiation",
142 context={"url": url, "formats_tried": list(PARSE_FORMATS)})
143 self.graph = graph
145 # Error (Fallback): Cannot Load RDF
146 except requests.RequestException as e:
147 raise ArtefactNotFoundError(
148 "Network error fetching artefact",
149 context={"url": url, "original_error": str(e)}
150 )
152 # ----------------------------------------------------------
153 # LOCAL FILE LOADING
154 # ----------------------------------------------------------
155 def _load_from_local_file(self, path: str) -> None:
156 with open(path, "rb") as f:
157 raw = f.read()
158 graph = parse_any(raw)
159 if graph is None:
160 raise ArtefactLoadError("Could not parse RDF with any known format",
161 context={"path": path, "formats_tried": list(PARSE_FORMATS)})
162 self.graph = graph
164 # ----------------------------------------------------------
165 # HELPERS
166 # ----------------------------------------------------------
167 def _is_url(self, s: str) -> bool:
168 try:
169 return urlparse(s).scheme in ("http", "https")
170 except Exception:
171 return False
173 def _guess_format_from_content_type(self, content_type: str) -> Optional[str]:
174 if "text/turtle" in content_type or "application/x-turtle" in content_type:
175 return "turtle"
176 if "application/rdf+xml" in content_type:
177 return "xml"
178 if "application/ld+json" in content_type or "json" in content_type:
179 return "json-ld"
180 if "application/n-triples" in content_type:
181 return "nt"
182 return None
184 def get_graph(self) -> Graph:
185 return self.graph
187 def _fetch_following_redirects(self, url: str, headers: dict, max_redirects: int = security.MAX_REDIRECTS):
188 """Follow URLs redirects manually, validating each hop via security.check_url_safe. Returns final response.
190 The default hop limit comes from security.MAX_REDIRECTS (env LODE_MAX_REDIRECTS)."""
191 current = url
192 for _ in range(max_redirects):
193 security.check_url_safe(current)
194 response = requests.get(current, headers=headers, timeout=10,
195 stream=True, allow_redirects=False)
196 # Validate the IP we ACTUALLY connected to, before reading anything:
197 # defeats DNS rebinding between check_url_safe above and this connect.
198 self._verify_peer_ip(response)
199 if response.status_code in (301, 302, 303, 307, 308):
200 location = response.headers.get("Location")
201 response.close()
202 if not location:
203 raise ArtefactLoadError("Redirect without Location", context={"url": current})
204 current = urljoin(current, location)
205 continue
206 return response
207 raise ArtefactLoadError("Too many redirects", context={"url": url})
209 def _verify_peer_ip(self, response) -> None:
210 """Re-check the SSRF policy against the socket's real peer IP. If it
211 cannot be determined (e.g. mocked in tests) fall back to the per-hop
212 check_url_safe already done before connecting."""
213 ip = self._peer_ip(response)
214 if ip is None:
215 return
216 try:
217 security.check_ip_safe(ip)
218 except ArtefactValidationError:
219 response.close()
220 raise
222 @staticmethod
223 def _peer_ip(response):
224 """Best-effort extraction of the connected peer IP from a streamed
225 requests response (reaches into urllib3 internals, hence defensive)."""
226 try:
227 return response.raw._connection.sock.getpeername()[0]
228 except Exception:
229 return None