Skip to content

Instantly share code, notes, and snippets.

@lastforkbender

lastforkbender/RDF_TYX_GENEITY.PY

Created June 14, 2026 14:04
Show Gist options

  • Select an option

    Learn more about clone URLs
  • Save lastforkbender/5da93f4622e38915f631c82ab43138f9 to your computer and use it in GitHub Desktop.

Select an option

Learn more about clone URLs
Save lastforkbender/5da93f4622e38915f631c82ab43138f9 to your computer and use it in GitHub Desktop.
Download ZIP
Extract, frame, canonicalize, sign, encrypt filename, wrap keys, emit turtle
Raw
RDF_TYX_GENEITY.PY
#!/usr/bin/env python3
"""
•••••••••••••••••••••••••••••••••••••••••••••••••••••••••••••••••••••••••••••••••••••••••••••••••••••
•••••••••••••••••••••••••••••••••••••••••••••••••••••••••••••••••••••••••••••••••••••••••••••••••••
RDF_TYX_GENEITY.PY
>>> Extract, frame, canonicalize, sign, encrypt filename, wrap keys, emit turtle
*Comprehensive logging, signature registry, js library inventory, verifiable data tracking
______________________________________________________________________________________________
-- NOTICE --
In advent of changes to this software, improper uses or other unexpected expectations. Thereof
herein no conditions are liable of that extent and deemed nonproprietary to public use further.
These no conditions not liable; remain intact regardless of any agencies demands. This software
isn't of any origins to incorporate any legislation effective to proprietary system protection.
The concepts & representing reach of this software intends criminal pattern analysis. Pertains
to any incompatible or compatible instances of use therein. You may freely distribute it's
associated coding yet there is absolutely no guarantee of any conditions clarifying proper
defined uses. These terms and otherwise conditions relevant or not per usage fully understood.
•••••••••••••••••••••••••••••••••••••••••••••••••••••••••••••••••••••••••••••••••••••••••••••••••••
•••••••••••••••••••••••••••••••••••••••••••••••••••••••••••••••••••••••••••••••••••••••••••••••••••••
"""
import os, json, base64, hashlib, argparse, logging, binascii
from pathlib import Path
from datetime import datetime
from typing import Dict, List, Tuple, Optional, Any
import requests
from w3lib.html import get_base_url
import extruct
from bs4 import BeautifulSoup
from rdflib import Graph, URIRef, Literal, Namespace
from rdflib.namespace import RDF, XSD
from pyld import jsonld
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey, Ed25519PublicKey
from cryptography.hazmat.primitives.asymmetric import rsa, padding
from cryptography.hazmat.primitives import serialization, hashes
from cryptography.hazmat.backends import default_backend
import cryptography.exceptions
def setup_logging(log_dir: Path = Path("./logs")):
# >>> Initialize multi-channel logging for application, signatures, and libraries
log_dir.mkdir(exist_ok=True)
timestamp = datetime.utcnow().strftime("%Y%m%d_%H%M%S")
app_log_file = log_dir / f"rdf_signal_{timestamp}.log"
sig_log_file = log_dir / f"signature_registry_{timestamp}.json"
lib_log_file = log_dir / f"js_library_inventory_{timestamp}.json"
logging.basicConfig(
level=logging.DEBUG,
format='%(asctime)s [%(levelname)s] %(name)s: %(message)s',
handlers=[
logging.FileHandler(app_log_file),
logging.StreamHandler() ] )
logger_instance = logging.getLogger(__name__)
logger_instance.info(f"Logging initialized. Log dir: {log_dir}")
return app_log_file, sig_log_file, lib_log_file
app_log_file, sig_log_file, lib_log_file = setup_logging()
logger = logging.getLogger(__name__)
# >>> Configuration <<<
KEY_DIR = Path("./keys")
KEY_DIR.mkdir(exist_ok=True)
PRIVATE_KEY_FILE = KEY_DIR / "ed25519_private.pem"
PUBLIC_KEY_FILE = KEY_DIR / "ed25519_public.pem"
PRIVATE_KEY_PASSPHRASE_ENV = "RDF_KEY_PW"
EX = Namespace("https://example.org/vocab#")
DCT = Namespace("http://purl.org/dc/terms/")
PROV = Namespace("http://www.w3.org/ns/prov#")
class SignatureRegistry:
# >>> Tracks all cryptographic signatures with full provenance and verification metadata
def __init__(self, log_file: Path):
self.log_file = log_file
self.entries: List[Dict] = []
self.sig_counter = 0
self.verification_map: Dict[str, Dict] = {}
def generate_static_id(self, prefix: str = "SIG") -> str:
# >>> Generate deterministic, static signature identifiers
self.sig_counter += 1
date_str = datetime.utcnow().strftime('%Y%m%d')
return f"{prefix}-{date_str}-{self.sig_counter:08d}"
def log_signature_creation(self,
data_hash: str,
algorithm: str,
key_id: str,
signature_b64: str,
data_preview: str,
data_size_bytes: int,
data_type: str = "nquads") -> str:
# >>> Log signature creation with full context
sig_id = self.generate_static_id("SIG")
entry = {
"timestamp": datetime.utcnow().isoformat() + "Z",
"event": "signature_created",
"signature_id": sig_id,
"algorithm": algorithm,
"key_id": key_id,
"data": {
"hash": data_hash,
"type": data_type,
"preview": data_preview[:300],
"size_bytes": data_size_bytes},
"signature_b64": signature_b64,
"cryptographic_context": {
"algorithm_family": "EdDSA",
"curve": "Ed25519",
"format": "LD-SIG:v1:alg=Ed25519;sig=<b64>",
"encoding": "base64" }}
self.entries.append(entry)
self.verification_map[sig_id] = {"created": entry, "verifications": []}
logger.info(f"• Signature created: {sig_id}")
return sig_id
def log_signature_verification(self, sig_id: str,
result: bool,
key_id: str = None,
error: str = None) -> None:
# >>> Log signature verification attempt
if sig_id not in self.verification_map:
logger.warning(f"Verification logged for unknown signature: {sig_id}")
self.verification_map[sig_id] = {"created": None, "verifications": []}
verification = {
"timestamp": datetime.utcnow().isoformat() + "Z",
"signature_id": sig_id,
"result": "VALID" if result else "INVALID",
"key_id": key_id,
"error": error}
self.verification_map[sig_id]["verifications"].append(verification)
status_icon = "✓" if result else "✗"
logger.info(f"{status_icon} Signature verification: {sig_id} = {verification['result']}")
def flush_to_file(self) -> None:
# >>> Write signature registry to JSON file
output = {
"generated_at": datetime.utcnow().isoformat() + "Z",
"total_signatures": len(self.entries),
"signatures": self.entries,
"verification_summary": {
sig_id: {
"created": data["created"]["timestamp"],
"verifications_count": len(data["verifications"]),
"all_valid": all(v["result"] == "VALID" for v in data["verifications"]) if data["verifications"] else None,
"verification_history": data["verifications"]}
for sig_id, data in self.verification_map.items()}}
try:
self.log_file.write_text(json.dumps(output, indent=2))
logger.info(f"• Signature registry flushed to {self.log_file}")
except Exception as e:
logger.error(f"Failed to write signature registry: {e}")
class JSLibraryInventory:
# >>> Tracks js libraries found in html, their versions and integrity attributes
def __init__(self, log_file: Path):
self.log_file = log_file
self.libraries: List[Dict] = []
self.sri_hashes: Dict[str, str] = {}
def add_script(self, src: str,
version: Optional[str] = None,
integrity_hash: Optional[str] = None,
sri_algorithm: Optional[str] = None,
crossorigin: Optional[str] = None,
context_url: str = None) -> None:
# >>> Register a discovered script/library
entry = {
"timestamp": datetime.utcnow().isoformat() + "Z",
"src": src,
"version": version,
"integrity": {
"hash": integrity_hash,
"algorithm": sri_algorithm
} if integrity_hash else None,
"crossorigin": crossorigin,
"discovered_at": context_url}
self.libraries.append(entry)
if integrity_hash:
self.sri_hashes[src] = integrity_hash
logger.debug(f"JS Library registered: {src} v{version or 'unknown'}")
def extract_from_html(self, html: str, base_url: str) -> None:
# >>> Parse HTML and extract all script tags
soup = BeautifulSoup(html, 'html.parser')
scripts = soup.find_all('script', src=True)
for script in scripts:
src = script.get('src', '').strip()
if not src:
continue
# >>> Resolve relative URLs
full_src = requests.compat.urljoin(base_url, src)
integrity = script.get('integrity', None)
crossorigin = script.get('crossorigin', None)
# >>> Try to extract version from URL
version = None
for pattern in [r'@(\d+\.\d+\.\d+)', r'/(\d+\.\d+\.\d+)/', r'v(\d+\.\d+\.\d+)']:
import re
match = re.search(pattern, src)
if match:
version = match.group(1)
break
# >>> Determine SRI algorithm if present
sri_algo = None
if integrity:
sri_algo = integrity.split('-')[0] if '-' in integrity else "sha256"
self.add_script(full_src, version=version, integrity_hash=integrity,
sri_algorithm=sri_algo, crossorigin=crossorigin,
context_url=base_url)
def flush_to_file(self) -> None:
# >>> Write library inventory to JSON file
output = {
"generated_at": datetime.utcnow().isoformat() + "Z",
"total_libraries": len(self.libraries),
"libraries": self.libraries,
"sri_coverage": {
"with_sri": len([l for l in self.libraries if l["integrity"]]),
"without_sri": len([l for l in self.libraries if not l["integrity"]]),
"sri_hashes": self.sri_hashes}}
try:
self.log_file.write_text(json.dumps(output, indent=2))
logger.info(f"• JS Library inventory flushed to {self.log_file}")
except Exception as e:
logger.error(f"Failed to write library inventory: {e}")
class VerifiableDataTracker:
# >>> Tracks all data inputs and outputs with hashes and provenance
def __init__(self):
self.data_map: Dict[str, Dict] = {}
self.static_id_counter = 0
def generate_data_id(self, prefix: str = "DATA") -> str:
# >>> Generate static data identifiers
self.static_id_counter += 1
date_str = datetime.utcnow().strftime('%Y%m%d')
return f"{prefix}-{date_str}-{self.static_id_counter:08d}"
def track_data(self, data: Any,
data_type: str,
source: str,
processing_stage: str) -> str:
# >>> Register data with provenance tracking
data_id = self.generate_data_id("DATA")
# >>> Compute hash based on data type
if isinstance(data, str):
data_bytes = data.encode('utf-8')
elif isinstance(data, bytes):
data_bytes = data
elif isinstance(data, dict):
data_bytes = json.dumps(data, sort_keys=True).encode('utf-8')
else:
data_bytes = str(data).encode('utf-8')
data_hash = "sha256:" + hashlib.sha256(data_bytes).hexdigest()
entry = {
"data_id": data_id,
"timestamp": datetime.utcnow().isoformat() + "Z",
"type": data_type,
"source": source,
"processing_stage": processing_stage,
"hash": data_hash,
"size_bytes": len(data_bytes),
"preview": str(data)[:500] if not isinstance(data, (dict, list)) else "object"}
self.data_map[data_id] = entry
logger.debug(f"Data tracked: {data_id} ({data_type})")
return data_id
def ensure_ed25519_keypair(passphrase: bytes = None) -> Tuple[Ed25519PrivateKey, Ed25519PublicKey]:
# >>> Generate or load Ed25519 keypair with enhanced error handling
if PRIVATE_KEY_FILE.exists() and PUBLIC_KEY_FILE.exists():
try:
with open(PRIVATE_KEY_FILE, "rb") as f:
pem = f.read()
priv = serialization.load_pem_private_key(pem, password=passphrase, backend=default_backend())
with open(PUBLIC_KEY_FILE, "rb") as f:
pub = serialization.load_pem_public_key(f.read(), backend=default_backend())
logger.info(f"• Loaded existing Ed25519 keypair from {KEY_DIR}")
return priv, pub
except ValueError as e:
logger.error(f"Failed to load private key: passphrase mismatch or corrupted key: {e}")
raise ValueError(f"Key loading failed: {e}")
except Exception as e:
logger.error(f"Unexpected error loading keys: {e}")
raise
logger.info("Generating new Ed25519 keypair...")
priv = Ed25519PrivateKey.generate()
pub = priv.public_key()
enc = serialization.BestAvailableEncryption(passphrase) if passphrase else serialization.NoEncryption()
priv_pem = priv.private_bytes(encoding=serialization.Encoding.PEM,
format=serialization.PrivateFormat.PKCS8,
encryption_algorithm=enc)
pub_pem = pub.public_bytes(encoding=serialization.Encoding.PEM,
format=serialization.PublicFormat.SubjectPublicKeyInfo)
PRIVATE_KEY_FILE.write_bytes(priv_pem)
PUBLIC_KEY_FILE.write_bytes(pub_pem)
logger.info(f"• Generated and saved Ed25519 keypair to {KEY_DIR}")
return priv, pub
def load_rsa_pub_pem(path: str) -> bytes:
# >>> Load and validate RSA public key PEM
try:
pem_data = Path(path).read_bytes()
# >>> Validate it's a valid RSA public key
pub = serialization.load_pem_public_key(pem_data, backend=default_backend())
if not isinstance(pub, rsa.RSAPublicKey):
raise ValueError("Public key is not RSA format")
logger.debug(f"• Valid RSA public key loaded from {path}")
return pem_data
except FileNotFoundError:
logger.error(f"RSA public key file not found: {path}")
raise
except Exception as e:
logger.error(f"Invalid RSA public key PEM at {path}: {e}")
raise ValueError(f"Invalid RSA public key: {e}")
def load_rsa_priv_pem(path: str, passphrase: bytes = None) -> Tuple[bytes, Optional[bytes]]:
# >>> Load and validate RSA private key PEM
try:
pem_data = Path(path).read_bytes()
priv = serialization.load_pem_private_key(pem_data, password=passphrase, backend=default_backend())
if not isinstance(priv, rsa.RSAPrivateKey):
raise ValueError("Private key is not RSA format")
logger.debug(f"• Valid RSA private key loaded from {path}")
return pem_data, passphrase
except FileNotFoundError:
logger.error(f"RSA private key file not found: {path}")
raise
except ValueError as e:
if "password" in str(e).lower():
logger.error(f"RSA private key passphrase mismatch at {path}")
raise ValueError(f"Invalid RSA private key: {e}")
except Exception as e:
logger.error(f"Error loading RSA private key from {path}: {e}")
raise
def fetch_html(url: str, timeout: int = 15) -> Tuple[str, str]:
# >>> Fetch HTML with robust error handling
try:
logger.info(f"Fetching URL: {url}")
r = requests.get(url, timeout=timeout)
r.raise_for_status()
logger.info(f"• Successfully fetched {url} ({len(r.text)} bytes)")
return r.text, get_base_url(r.text, r.url)
except requests.exceptions.Timeout:
logger.error(f"Timeout fetching {url} (>{timeout}s)")
raise
except requests.exceptions.HTTPError as e:
logger.error(f"HTTP error {e.response.status_code} from {url}")
raise
except requests.exceptions.ConnectionError as e:
logger.error(f"Connection error fetching {url}: {e}")
raise
except Exception as e:
logger.error(f"Unexpected error fetching {url}: {type(e).__name__}: {e}")
raise
def extract_rdf_from_html(html: str, base: str) -> Dict:
# >>> Extract RDF with error tracking
logger.info("Extracting RDF (JSON-LD, RDFa, Microdata)...")
try:
extracted = extruct.extract(html, base_url=base, syntaxes=['json-ld', 'rdfa', 'microdata'])
logger.info(f"• Extraction complete: json-ld={len(extracted.get('json-ld',[]))}, "
f"rdfa={len(extracted.get('rdfa',[]))}, microdata={len(extracted.get('microdata',[]))}")
return extracted
except Exception as e:
logger.error(f"RDF extraction failed: {e}")
raise
def find_linked_rdf_urls(html: str, base: str) -> List[Tuple[str, str]]:
# >>> Find linked RDF resources with validation
logger.info("Scanning for linked RDF resources...")
soup = BeautifulSoup(html, 'html.parser')
out = []
rdf_mimetypes = {
'application/ld+json', 'application/rdf+xml',
'text/turtle', 'application/n-triples', 'text/n-triples'}
for link in soup.find_all('link', href=True):
mimetype = link.get('type', '').lower().strip()
if mimetype in rdf_mimetypes:
href = requests.compat.urljoin(base, link['href'])
out.append((href, mimetype))
logger.debug(f"Found linked RDF: {href} ({mimetype})")
logger.info(f"• Found {len(out)} linked RDF resource(s)")
return out
def parse_rdf_string(payload: str, fmt_hint: str = None) -> Graph:
# >>> Parse RDF with format detection and error handling
g = Graph()
fmt_map = {
'application/ld+json': 'json-ld',
'json-ld': 'json-ld',
'text/turtle': 'turtle',
'turtle': 'turtle',
'application/rdf+xml': 'xml',
'application/n-triples': 'nt',
'text/n-triples': 'nt'}
fmt = fmt_map.get(fmt_hint, None)
if fmt:
try:
g.parse(data=payload, format=fmt)
logger.debug(f"• Parsed RDF as {fmt} (hint-based)")
return g
except Exception as e:
logger.warning(f"Format hint {fmt_hint} failed: {e}, attempting auto-detection...")
for try_fmt in ('json-ld', 'turtle', 'xml', 'nt'):
try:
g.parse(data=payload, format=try_fmt)
logger.debug(f"• Parsed RDF as {try_fmt} (auto-detected)")
return g
except Exception:
continue
logger.error(f"Could not parse RDF payload, tried all formats")
raise ValueError("Could not parse RDF payload with any supported format")
def frame_jsonld_for_signal(jsonobj: Dict) -> Tuple[Dict, List]:
# >>> Frame JSON-LD for Signal extraction with error handling
try:
frame = {
"@context": {
"ex": str(EX),
"prov": str(PROV),
"dct": str(DCT)},
"@type": "ex:Signal",
"@embed": "@always"}
framed = jsonld.frame(jsonobj, frame)
graph = framed.get('@graph') or ([framed] if '@type' in framed else [])
logger.info(f"• Framed {len(graph)} Signal object(s)")
return framed, graph
except Exception as e:
logger.warning(f"JSON-LD framing failed: {e}")
return jsonobj, []
def graph_to_canonical_nquads(graph: Graph) -> str:
# >>> Convert RDF Graph to canonical N-Quads with error handling
try:
jld = graph.serialize(format='json-ld', indent=2)
jobj = json.loads(jld)
nquads = jsonld.normalize(jobj, {'algorithm': 'URDNA2015', 'format': 'application/n-quads'})
logger.info(f"• Canonicalized to {len(nquads)} N-Quads bytes")
return nquads
except Exception as e:
logger.error(f"Graph serialization/canonicalization failed: {e}")
raise ValueError(f"Cannot canonicalize RDF graph: {e}")
def sign_nquads(nquads: str, priv_key: Ed25519PrivateKey, sig_registry: SignatureRegistry, key_id: str = "default") -> Tuple[str, str]:
# >>> Sign N-Quads and log to signature registry
try:
sig = priv_key.sign(nquads.encode('utf-8'))
sig_b64 = base64.b64encode(sig).decode('utf-8')
sig_block = "LD-SIG:v1:alg=Ed25519;sig=" + sig_b64
data_hash = "sha256:" + hashlib.sha256(nquads.encode('utf-8')).hexdigest()
sig_id = sig_registry.log_signature_creation(
data_hash=data_hash,
algorithm="Ed25519",
key_id=key_id,
signature_b64=sig_b64,
data_preview=nquads[:300],
data_size_bytes=len(nquads),
data_type="canonicalized_nquads")
logger.info(f"• Signed N-Quads with Ed25519 (signature ID: {sig_id})")
return sig_block, sig_id
except Exception as e:
logger.error(f"Signing failed: {e}")
raise
def verify_nquads_signature(nquads: str, sig_block: str,
pub_key: Ed25519PublicKey,
sig_registry: SignatureRegistry,
sig_id: str = None) -> bool:
# >>> Verify N-Quads signature with logging
try:
if not sig_block.startswith("LD-SIG:v1:alg=Ed25519;sig="):
logger.error(f"Invalid signature format: {sig_block[:50]}...")
if sig_registry and sig_id:
sig_registry.log_signature_verification(sig_id, False, error="Invalid format")
return False
b64 = sig_block.split("sig=")[1]
sig = base64.b64decode(b64.encode('utf-8'))
pub_key.verify(sig, nquads.encode('utf-8'))
logger.info(f"• Signature verification PASSED")
if sig_registry and sig_id:
sig_registry.log_signature_verification(sig_id, True)
return True
except cryptography.exceptions.InvalidSignature:
logger.warning(f"Signature verification FAILED: Invalid signature")
if sig_registry and sig_id:
sig_registry.log_signature_verification(sig_id, False, error="Invalid signature")
return False
except (ValueError, binascii.Error) as e:
logger.error(f"Signature verification error (decoding): {e}")
if sig_registry and sig_id:
sig_registry.log_signature_verification(sig_id, False, error=f"Decoding error: {e}")
return False
except Exception as e:
logger.error(f"Unexpected error during signature verification: {e}")
if sig_registry and sig_id:
sig_registry.log_signature_verification(sig_id, False, error=f"Unexpected: {e}")
raise
def encrypt_filename_block(plaintext_obj: Dict, key: bytes) -> str:
# >>> Encrypt filename metadata block with error handling
try:
aesgcm = AESGCM(key)
iv = os.urandom(12)
pdata = json.dumps(plaintext_obj).encode('utf-8')
ct = aesgcm.encrypt(iv, pdata, None)
block = {
"v": "1",
"alg": "AES-GCM",
"iv": base64.b64encode(iv).decode('utf-8'),
"ct": base64.b64encode(ct).decode('utf-8')}
b64 = base64.b64encode(json.dumps(block).encode('utf-8')).decode('utf-8')
encrypted_block = f"ENC:{b64}"
logger.debug(f"• Encrypted filename block ({len(pdata)} bytes plaintext)")
return encrypted_block
except Exception as e:
logger.error(f"Filename encryption failed: {e}")
raise ValueError(f"Encryption failed: {e}")
def decrypt_filename_block(block_str: str, key: bytes) -> Dict:
# >>> Decrypt filename metadata block with full error handling
try:
if not block_str.startswith("ENC:"):
raise ValueError("Invalid encrypted block format")
b64 = block_str[len("ENC:"):]
raw = base64.b64decode(b64.encode('utf-8'))
obj = json.loads(raw.decode('utf-8'))
iv = base64.b64decode(obj['iv'])
ct = base64.b64decode(obj['ct'])
aesgcm = AESGCM(key)
pt = aesgcm.decrypt(iv, ct, None)
result = json.loads(pt.decode('utf-8'))
logger.debug(f"• Decrypted filename block")
return result
except (ValueError, binascii.Error) as e:
logger.error(f"Invalid base64 in encrypted block: {e}")
raise ValueError(f"Base64 decoding failed: {e}")
except Exception as e:
logger.error(f"Filename decryption failed: {e}")
raise ValueError(f"Decryption failed: {e}")
def wrap_key_rsa_oaep(sym_key: bytes, recipient_pub_pem: bytes, recipient_id: str = "unknown") -> str:
# >>> Wrap symmetric key with RSA-OAEP with full validation
try:
pub = serialization.load_pem_public_key(recipient_pub_pem, backend=default_backend())
if not isinstance(pub, rsa.RSAPublicKey):
raise ValueError(f"Public key for {recipient_id} is not RSA format")
ct = pub.encrypt(sym_key, padding.OAEP(
mgf=padding.MGF1(algorithm=hashes.SHA256()),
algorithm=hashes.SHA256(),
label=None))
wrapped = "KW-RSA-OAEP:v1:" + base64.b64encode(ct).decode('utf-8')
logger.debug(f"• Wrapped symmetric key for {recipient_id}")
return wrapped
except TypeError:
logger.error(f"Failed to load RSA public key for {recipient_id}")
raise ValueError(f"Invalid RSA public key for {recipient_id}")
except Exception as e:
logger.error(f"Key wrapping failed for {recipient_id}: {e}")
raise ValueError(f"Key wrapping failed: {e}")
def unwrap_key_rsa_oaep(wrapped_str: str, recipient_priv_pem: bytes, passphrase: bytes = None) -> bytes:
# >>> Unwrap RSA-encrypted symmetric key with error handling
try:
if not wrapped_str.startswith("KW-RSA-OAEP:v1:"):
raise ValueError("Invalid key wrapping format")
ct = base64.b64decode(wrapped_str.split(":", 2)[2])
priv = serialization.load_pem_private_key(recipient_priv_pem, password=passphrase, backend=default_backend())
if not isinstance(priv, rsa.RSAPrivateKey):
raise ValueError("Private key is not RSA format")
key = priv.decrypt(ct, padding.OAEP(
mgf=padding.MGF1(algorithm=hashes.SHA256()),
algorithm=hashes.SHA256(),
label=None))
logger.debug(f"• Unwrapped symmetric key")
return key
except (ValueError, binascii.Error) as e:
logger.error(f"Key unwrapping failed: {e}")
raise ValueError(f"Key unwrapping failed: {e}")
except Exception as e:
logger.error(f"Unexpected error unwrapping key: {e}")
raise
def build_signal_turtle(signal_uri: str,
title: str,
created_iso: str,
content_hash: str,
enc_fname_block: str,
signature: str,
sig_id: str,
wrapped_keys: List[Dict],
creator_uri: str = None) -> str:
# >>> Build Turtle RDF Signal with enhanced metadata
try:
g = Graph()
g.bind("ex", EX)
g.bind("dct", DCT)
g.bind("prov", PROV)
s = URIRef(signal_uri)
g.add((s, RDF.type, EX.Signal))
g.add((s, DCT.title, Literal(title)))
g.add((s, DCT.created, Literal(created_iso, datatype=XSD.dateTime)))
g.add((s, EX.contentHash, Literal(content_hash)))
g.add((s, EX.encryptedFilenameBlock, Literal(enc_fname_block)))
g.add((s, EX.signature, Literal(signature)))
g.add((s, EX.signatureId, Literal(sig_id)))
if creator_uri:
g.add((s, DCT.creator, URIRef(creator_uri)))
# >>> Add wrapped keys as blank nodes with full metadata
for idx, wk in enumerate(wrapped_keys):
wk_hash = hashlib.sha1(wk['wrapped'].encode()).hexdigest()[:8]
bn = URIRef(f"{signal_uri}#wrappedKey-{idx}-{wk_hash}")
g.add((s, EX.wrappedKey, bn))
g.add((bn, EX.keyFor, Literal(wk['recipient'])))
g.add((bn, EX.wrappedValue, Literal(wk['wrapped'])))
g.add((bn, EX.wrappingAlgorithm, Literal("RSA-OAEP")))
ttl = g.serialize(format='turtle')
result = ttl.decode('utf-8') if isinstance(ttl, bytes) else ttl
logger.info(f"• Built Turtle Signal with {len(wrapped_keys)} wrapped key(s)")
return result
except Exception as e:
logger.error(f"Failed to build Turtle Signal: {e}")
raise ValueError(f"Signal construction failed: {e}")
def process_url_full(url: str,
recipient_pub_pem_paths: List[str],
rsa_priv_unwrap_path: str = None,
rsa_priv_passphrase: bytes = None,
signer_passphrase: bytes = None) -> Optional[Dict]:
logger.info("=" * 80)
logger.info(f"Starting RDF Signal Pipeline for: {url}")
logger.info("=" * 80)
# >>> Initialize tracking systems
sig_registry = SignatureRegistry(sig_log_file)
js_inventory = JSLibraryInventory(lib_log_file)
data_tracker = VerifiableDataTracker()
try:
# >>> Ensure signer keys
logger.info("\n[STEP 1] Initializing Ed25519 signer keypair...")
signer_priv, signer_pub = ensure_ed25519_keypair(passphrase=signer_passphrase)
signer_key_id = "default-ed25519"
# >>> Fetch HTML
logger.info("\n[STEP 2] Fetching HTML content...")
html, base = fetch_html(url)
data_tracker.track_data(html, "html_document", url, "fetched")
# >>> Extract JS libraries
logger.info("\n[STEP 3] Extracting JavaScript libraries...")
js_inventory.extract_from_html(html, base)
# >>> Extract RDF
logger.info("\n[STEP 4] Extracting RDF data...")
extracted = extract_rdf_from_html(html, base)
combined_graph = Graph()
# >>> Process JSON-LD from extruct
jsonld_count = 0
for jd in extracted.get('json-ld', []):
try:
combined_graph.parse(data=json.dumps(jd), format='json-ld')
jsonld_count += 1
except Exception as e:
logger.warning(f"Failed to parse JSON-LD: {e}")
continue
logger.info(f"• Processed {jsonld_count} JSON-LD object(s)")
# >>> Process RDFa from extruct
rdfa_count = 0
for rdfa in extracted.get('rdfa', []):
try:
# >>> Handle both string and dict RDFa
if isinstance(rdfa, dict):
rdfa_str = json.dumps(rdfa)
elif isinstance(rdfa, str):
rdfa_str = rdfa
else:
logger.warning(f"Skipping RDFa of unexpected type: {type(rdfa)}")
continue
combined_graph.parse(data=rdfa_str, format='rdfa')
rdfa_count += 1
except Exception as e:
logger.warning(f"Failed to parse RDFa: {e}")
continue
logger.info(f"• Processed {rdfa_count} RDFa block(s)")
# >>> Parse linked RDF files
logger.info("\nFetching linked RDF resources...")
linked_urls = find_linked_rdf_urls(html, base)
linked_count = 0
for href, mime in linked_urls:
try:
r = requests.get(href, timeout=10)
r.raise_for_status()
content = r.content.decode('utf-8', errors='ignore')
g = parse_rdf_string(content, fmt_hint=mime)
for t in g:
combined_graph.add(t)
linked_count += 1
logger.info(f"• Processed linked RDF from {href}")
except requests.exceptions.Timeout:
logger.warning(f"Timeout fetching linked RDF: {href}")
except requests.exceptions.HTTPError as e:
logger.warning(f"HTTP {e.response.status_code} from {href}")
except Exception as e:
logger.warning(f"Error processing linked RDF {href}: {type(e).__name__}")
logger.info(f"• Processed {linked_count} linked RDF resource(s)")
# >>> Validate extracted RDF
logger.info("\nValidating extracted RDF...")
if len(combined_graph) == 0:
logger.warning(f"No RDF triples extracted. Summary: "
f"json-ld={jsonld_count}, rdfa={rdfa_count}, linked={linked_count}")
return None
logger.info(f"• Combined graph has {len(combined_graph)} triples")
# >>> Frame JSON-LD
logger.info("\nFraming JSON-LD...")
jld = combined_graph.serialize(format='json-ld', indent=2)
jobj = json.loads(jld)
framed, signals = frame_jsonld_for_signal(jobj)
# >>> Canonicalize and sign
logger.info("\nCanonicalizing RDF...")
nquads = graph_to_canonical_nquads(combined_graph)
data_tracker.track_data(nquads, "canonical_nquads", "canonicalization", "processed")
content_hash = "sha256:" + hashlib.sha256(nquads.encode('utf-8')).hexdigest()
logger.info(f"• Content hash: {content_hash}")
logger.info("\n[STEP 9] Signing canonical RDF...")
signature, sig_id = sign_nquads(nquads, signer_priv, sig_registry, key_id=signer_key_id)
# >>> Encrypt filename metadata
logger.info("\nCreating encrypted filename block...")
sym_key = AESGCM.generate_key(bit_length=256)
filename_meta = {
"filename": "extracted_signal.csv",
"source_url": url,
"extracted_at": datetime.utcnow().isoformat() + "Z",
"triple_count": len(combined_graph)}
enc_fname_block = encrypt_filename_block(filename_meta, sym_key)
logger.info(f"• Encrypted filename block (metadata: {list(filename_meta.keys())})")
# >>> Wrap symmetric key for recipients
logger.info("\nWrapping symmetric key for recipients...")
wrapped_keys = []
for pub_path in recipient_pub_pem_paths:
try:
recipient_id = Path(pub_path).stem
recipient_pub_pem = load_rsa_pub_pem(pub_path)
wrapped = wrap_key_rsa_oaep(sym_key, recipient_pub_pem, recipient_id=recipient_id)
wrapped_keys.append({"recipient": recipient_id, "wrapped": wrapped})
logger.info(f"• Wrapped key for recipient: {recipient_id}")
except Exception as e:
logger.error(f"Failed to wrap key for {pub_path}: {e}")
continue
if not wrapped_keys:
logger.error("No keys successfully wrapped")
return None
logger.info(f"• Successfully wrapped {len(wrapped_keys)} key(s)")
# >>> Build Turtle Signal
logger.info("\nBuilding Turtle Signal...")
signal_uri = f"urn:sha256:{hashlib.sha256((content_hash + enc_fname_block).encode()).hexdigest()}"
created_iso = datetime.utcnow().isoformat() + "Z"
ttl = build_signal_turtle(
signal_uri=signal_uri,
title="Extracted RDF Signal",
created_iso=created_iso,
content_hash=content_hash,
enc_fname_block=enc_fname_block,
signature=signature,
sig_id=sig_id,
wrapped_keys=wrapped_keys,
creator_uri=None)
# >>> Verify signature
logger.info("\nVerifying signature...")
sig_valid = verify_nquads_signature(nquads, signature, signer_pub, sig_registry, sig_id=sig_id)
logger.info(f"Signature verification: {'PASSED' if sig_valid else 'FAILED'}")
# >>> Flush logs
logger.info("\n[STEP 14] Flushing logs and registries...")
sig_registry.flush_to_file()
js_inventory.flush_to_file()
logger.info("\n" + "=" * 80)
logger.info("Pipeline completed successfully!")
logger.info("=" * 80)
return {
"turtle": ttl,
"signal_uri": signal_uri,
"framed_signals": framed,
"signature": signature,
"signature_id": sig_id,
"signature_valid": sig_valid,
"content_hash": content_hash,
"triple_count": len(combined_graph),
"wrapped_keys_count": len(wrapped_keys),
"js_libraries_found": len(js_inventory.libraries),
"extracted_at": created_iso,
"logs": {
"app_log": str(app_log_file),
"signature_registry": str(sig_log_file),
"js_inventory": str(lib_log_file)}}
except Exception as e:
logger.error(f"\nPipeline failed: {type(e).__name__}: {e}", exc_info=True)
return None
def main():
# >>> Command-line interface for RDF Signal pipeline
p = argparse.ArgumentParser(
description="Extract, canonicalize, sign, and encrypt RDF signals from URLs. "
"Produces Turtle output with signature tracking and JS library inventory.",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""
Examples:
python rdf_signal_full.py https://example.com/page \\
--recipients /path/to/recipient_pub.pem /path/to/another_pub.pem
RDF_KEY_PW="my-passphrase" python rdf_signal_full.py https://example.com \\
--recipients recipient.pem --signer-pass-env RDF_KEY_PW
""")
p.add_argument("url", help="URL to scan for RDF content")
p.add_argument("--recipients", nargs="+", required=True,
help="Paths to recipients' RSA public PEM files (required)")
p.add_argument("--signer-pass-env", default=PRIVATE_KEY_PASSPHRASE_ENV,
help=f"Environment variable name for signer private key passphrase "
f"(default: {PRIVATE_KEY_PASSPHRASE_ENV})")
p.add_argument("--output-dir", default="./output",
help="Directory for output files (default: ./output)")
p.add_argument("--log-dir", default="./logs",
help="Directory for log files (default: ./logs)")
args = p.parse_args()
# >>> Create output directory
output_dir = Path(args.output_dir)
output_dir.mkdir(exist_ok=True)
# >>> Get signer passphrase from environment if set
pass_env = os.environ.get(args.signer_pass_env)
pass_bytes = pass_env.encode('utf-8') if pass_env else None
if pass_env:
logger.info(f"Using signer passphrase from environment variable: {args.signer_pass_env}")
else:
logger.info(f"No signer passphrase set (environment variable: {args.signer_pass_env})")
# >>> Validate recipient files exist
for recipient in args.recipients:
if not Path(recipient).exists():
logger.error(f"Recipient public key file not found: {recipient}")
return 1
# >>> Run pipeline
logger.info(f"Processing URL: {args.url}")
logger.info(f"Recipients: {', '.join(Path(r).name for r in args.recipients)}")
result = process_url_full(args.url, args.recipients, signer_passphrase=pass_bytes)
if result is None:
logger.error("Pipeline produced no output")
return 1
# >>> Write Turtle output
ttl_file = output_dir / "signal_output.ttl"
ttl_file.write_text(result['turtle'])
logger.info(f"• Turtle Signal written to: {ttl_file}")
# >>> Write JSON summary
summary = {
"status": "success",
"completed_at": datetime.utcnow().isoformat() + "Z",
"source_url": args.url,
"signal_uri": result['signal_uri'],
"signatures": {
"signature_id": result['signature_id'],
"algorithm": "Ed25519",
"format": "LD-SIG:v1",
"verified": result['signature_valid']
},
"data": {
"content_hash": result['content_hash'],
"triple_count": result['triple_count'],
"wrapped_keys_count": result['wrapped_keys_count']
},
"extraction": {
"javascript_libraries": result['js_libraries_found'],
"extracted_at": result['extracted_at']
},
"files": {
"turtle_signal": str(ttl_file),
"app_log": result['logs']['app_log'],
"signature_registry": result['logs']['signature_registry'],
"js_inventory": result['logs']['js_inventory']}}
summary_file = output_dir / "summary.json"
summary_file.write_text(json.dumps(summary, indent=2))
logger.info(f"• Summary written to: {summary_file}")
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment