mengzhx/semantica
1
0
Fork 0
mirror of https://github.com/semantica-agi/semantica.git synced 2026-09-12 04:01:35 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(blazegraph): expand prefixed datatypes and validate lang/datatype metadata

Browse Source 
- Added _resolve_datatype_iri() to expand known prefixes (xsd/rdf/rdfs/owl/skos)
  to full IRIs instead of blindly wrapping in <...>, fixing invalid SPARQL like
  <xsd:integer>
- Validated language tags against RFC 5646 regex to prevent SPARQL injection
  via metadata["lang"] values containing whitespace or punctuation
- Validated datatype IRIs for whitespace/special characters before interpolation
- Extended test suite from 7 to 15 cases covering prefix expansion, injection
  rejection, and all accepted input forms
This commit is contained in:
KaifAhmad1
2026-04-11 15:16:15 +05:30
parent 3a1a798107
commit 2e2dae558f
2 changed files with 173 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
semantica/triplet_store/blazegraph_store.py
+61 -4
View File
@@ -27,6 +27,7 @@ Author: Semantica Contributors
License: MIT
"""
import re
from typing import Any, Dict, List, Optional
from urllib.parse import urljoin, urlparse
@@ -254,6 +255,18 @@ class BlazegraphStore:
)
return " ".join(lines)
# Known prefix expansions for XSD and common RDF vocabularies
_KNOWN_PREFIXES: Dict[str, str] = {
"xsd": "http://www.w3.org/2001/XMLSchema#",
"rdf": "http://www.w3.org/1999/02/22-rdf-syntax-ns#",
"rdfs": "http://www.w3.org/2000/01/rdf-schema#",
"owl": "http://www.w3.org/2002/07/owl#",
"skos": "http://www.w3.org/2004/02/skos/core#",
}
# RFC 5646 language tag: primary subtag optionally followed by '-' + subtags
_LANG_TAG_RE = re.compile(r"^[a-zA-Z]{1,8}(-[a-zA-Z0-9]{1,8})*$")
def _format_object_for_sparql(self, triplet: Triplet) -> str:
"""Format triplet object as IRI or literal for SPARQL/N-Triples style syntax."""
obj = triplet.object
@@ -261,6 +274,9 @@ class BlazegraphStore:
if self._is_uri_value(obj):
if obj.startswith("<") and obj.endswith(">"):
inner = obj[1:-1]
if " " in inner or ">" in inner:
raise ValueError(f"IRI contains invalid characters: {obj!r}")
return obj
return f"<{obj}>"
@@ -269,16 +285,54 @@ class BlazegraphStore:
language = metadata.get("lang") or metadata.get("language")
if datatype:
datatype_iri = datatype
if not datatype_iri.startswith("<"):
datatype_iri = f"<{datatype_iri}>"
datatype_iri = self._resolve_datatype_iri(datatype)
return f"\"{escaped}\"^^{datatype_iri}"
if language:
if not self._LANG_TAG_RE.match(str(language)):
raise ValueError(
f"Invalid language tag {language!r}: must match RFC 5646 "
f"(letters/digits and hyphens only, e.g. 'en', 'en-US')"
)
return f"\"{escaped}\"@{language}"
return f"\"{escaped}\""
def _resolve_datatype_iri(self, datatype: str) -> str:
"""Expand a datatype string to a validated SPARQL IRI token.
Accepts:
- Already-wrapped IRIs: ``<http://...>``
- Full IRIs: ``http://...`` / ``https://...`` / ``urn:...``
- Known prefixed names: ``xsd:integer``, ``rdf:langString``, etc.
Raises ValueError for anything else.
"""
datatype = str(datatype)
# Already angle-bracketed — validate the inner IRI contains no whitespace
if datatype.startswith("<") and datatype.endswith(">"):
inner = datatype[1:-1]
if not inner or re.search(r"[\s<>\"{}|\\^`]", inner):
raise ValueError(f"Invalid datatype IRI: {datatype!r}")
return datatype
# Full absolute IRI without brackets
parsed = urlparse(datatype)
if parsed.scheme in {"http", "https", "urn"} and not re.search(r"[\s<>\"{}|\\^`]", datatype):
return f"<{datatype}>"
# Prefixed form — expand known prefixes only
if ":" in datatype:
prefix, local = datatype.split(":", 1)
if prefix in self._KNOWN_PREFIXES and re.match(r"^[A-Za-z0-9_\-\.]+$", local):
return f"<{self._KNOWN_PREFIXES[prefix]}{local}>"
raise ValueError(
f"Unsupported datatype {datatype!r}: use a full IRI (http/https/urn), "
f"an angle-bracketed IRI, or a known prefix (xsd/rdf/rdfs/owl/skos)."
)
def _is_uri_value(self, value: str) -> bool:
"""Detect if a value should be serialized as an IRI."""
if not isinstance(value, str) or not value:
@@ -286,7 +340,10 @@ class BlazegraphStore:
if value.startswith("<") and value.endswith(">"):
return True
parsed = urlparse(value)
return parsed.scheme in {"http", "https", "urn"}
if parsed.scheme not in {"http", "https", "urn"}:
return False
# Reject strings that only look like URIs (e.g. "http not a uri")
return not re.search(r"\s", value)
def _escape_literal(self, value: str) -> str:
"""Escape string literal for SPARQL."""
tests/triplet_store/test_blazegraph_store.py
+112
View File
@@ -108,6 +108,118 @@ class TestBlazegraphStoreSerialization(unittest.TestCase):
obj = store._format_object_for_sparql(triplet)
self.assertEqual(obj, "\"http not a uri\"")
# --- Bug 1: prefixed datatype expansion ---
@patch.object(BlazegraphStore, "_connect", autospec=True)
def test_format_object_expands_xsd_prefix_to_full_iri(self, _mock_connect):
store = BlazegraphStore(endpoint="http://localhost:9999/blazegraph")
triplet = Triplet(
subject="urn:entity:person:1",
predicate="urn:property:age",
object="42",
metadata={"datatype": "xsd:integer"},
)
obj = store._format_object_for_sparql(triplet)
self.assertEqual(
obj,
"\"42\"^^<http://www.w3.org/2001/XMLSchema#integer>",
)
@patch.object(BlazegraphStore, "_connect", autospec=True)
def test_format_object_expands_rdf_prefix_to_full_iri(self, _mock_connect):
store = BlazegraphStore(endpoint="http://localhost:9999/blazegraph")
triplet = Triplet(
subject="urn:entity:person:1",
predicate="urn:property:value",
object="hello",
metadata={"datatype": "rdf:langString"},
)
obj = store._format_object_for_sparql(triplet)
self.assertEqual(
obj,
"\"hello\"^^<http://www.w3.org/1999/02/22-rdf-syntax-ns#langString>",
)
@patch.object(BlazegraphStore, "_connect", autospec=True)
def test_format_object_rejects_unknown_prefix(self, _mock_connect):
store = BlazegraphStore(endpoint="http://localhost:9999/blazegraph")
triplet = Triplet(
subject="urn:entity:person:1",
predicate="urn:property:value",
object="hello",
metadata={"datatype": "myns:customType"},
)
with self.assertRaises(ValueError):
store._format_object_for_sparql(triplet)
# --- Bug 2: metadata injection validation ---
@patch.object(BlazegraphStore, "_connect", autospec=True)
def test_format_object_rejects_injected_lang_tag(self, _mock_connect):
store = BlazegraphStore(endpoint="http://localhost:9999/blazegraph")
triplet = Triplet(
subject="urn:entity:person:1",
predicate="urn:property:label",
object="Color",
metadata={"lang": "en . CLEAR ALL #"},
)
with self.assertRaises(ValueError):
store._format_object_for_sparql(triplet)
@patch.object(BlazegraphStore, "_connect", autospec=True)
def test_format_object_rejects_datatype_with_whitespace(self, _mock_connect):
store = BlazegraphStore(endpoint="http://localhost:9999/blazegraph")
triplet = Triplet(
subject="urn:entity:person:1",
predicate="urn:property:age",
object="42",
metadata={"datatype": "http://example.org/type CLEAR ALL"},
)
with self.assertRaises(ValueError):
store._format_object_for_sparql(triplet)
@patch.object(BlazegraphStore, "_connect", autospec=True)
def test_format_object_accepts_full_iri_datatype_no_brackets(self, _mock_connect):
store = BlazegraphStore(endpoint="http://localhost:9999/blazegraph")
triplet = Triplet(
subject="urn:entity:person:1",
predicate="urn:property:age",
object="42",
metadata={"datatype": "http://www.w3.org/2001/XMLSchema#integer"},
)
obj = store._format_object_for_sparql(triplet)
self.assertEqual(
obj,
"\"42\"^^<http://www.w3.org/2001/XMLSchema#integer>",
)
@patch.object(BlazegraphStore, "_connect", autospec=True)
def test_format_object_accepts_bracketed_iri_datatype(self, _mock_connect):
store = BlazegraphStore(endpoint="http://localhost:9999/blazegraph")
triplet = Triplet(
subject="urn:entity:person:1",
predicate="urn:property:age",
object="42",
metadata={"datatype": "<http://www.w3.org/2001/XMLSchema#integer>"},
)
obj = store._format_object_for_sparql(triplet)
self.assertEqual(
obj,
"\"42\"^^<http://www.w3.org/2001/XMLSchema#integer>",
)
@patch.object(BlazegraphStore, "_connect", autospec=True)
def test_format_object_accepts_hyphenated_lang_tag(self, _mock_connect):
store = BlazegraphStore(endpoint="http://localhost:9999/blazegraph")
triplet = Triplet(
subject="urn:entity:person:1",
predicate="urn:property:label",
object="Colour",
metadata={"lang": "en-GB"},
)
obj = store._format_object_for_sparql(triplet)
self.assertEqual(obj, "\"Colour\"@en-GB")
if __name__ == "__main__":
unittest.main()