mirror of
https://github.com/safishamsi/graphify.git
synced 2026-09-25 06:55:43 +00:00
security: SSRF protection, HTML escaping, path guards, encoding hardening
- graphify/security.py (new): centralised security module
- validate_url(): blocks file://, ftp://, data:, any non-http/https scheme
- _NoFileRedirectHandler: re-validates redirect targets, blocks file:// redirects
- safe_fetch(): streams response, 50MB hard cap, non-2xx raises, timeout
- safe_fetch_text(): safe_fetch + UTF-8 decode with errors=replace
- validate_graph_path(): resolves path, requires inside .graphify/, base must exist
- sanitize_label(): strip control chars, cap 256, html.escape() — mirrors
code-review-graph's _sanitize_name pattern
- graphify/ingest.py: _fetch_html() and _download_binary() now use safe_fetch*;
ingest() validates URL scheme and wraps network calls in try/except;
YAML frontmatter: newlines stripped from question before embedding
- graphify/extract.py: all 33 bare .decode() → .decode("utf-8", errors="replace")
— non-UTF-8 source files degrade gracefully instead of crashing extraction
- graphify/export.py: sanitize_label() on all node labels and edge titles
before pyvis embeds them in HTML output
- graphify/serve.py: _load_graph() validates graph_path via validate_graph_path()
and wraps JSONDecodeError with recovery message; sanitize_label() on MCP
text output
- graphify/detect.py: os.walk(..., followlinks=False) made explicit
- SECURITY.md (new): threat model, mitigations table, reporting process
- tests/test_security.py (new): 20 tests covering all security.py functions
This commit is contained in:
Safi
+1
-1
@@ -156,7 +156,7 @@ def detect(root: Path) -> dict:
|
||||
for scan_root in scan_paths:
|
||||
in_memory_tree = memory_dir.exists() and str(scan_root).startswith(str(memory_dir))
|
||||
import os
|
||||
for dirpath, dirnames, filenames in os.walk(scan_root):
|
||||
for dirpath, dirnames, filenames in os.walk(scan_root, followlinks=False):
|
||||
dp = Path(dirpath)
|
||||
if not in_memory_tree:
|
||||
# Prune noise dirs in-place so os.walk never descends into them
|
||||
|
||||
+3
-2
@@ -7,6 +7,7 @@ from collections import Counter
|
||||
from pathlib import Path
|
||||
import networkx as nx
|
||||
from networkx.readwrite import json_graph
|
||||
from graphify.security import sanitize_label
|
||||
|
||||
COMMUNITY_COLORS = [
|
||||
"#4E79A7", "#F28E2B", "#E15759", "#76B7B2", "#59A14F",
|
||||
@@ -71,9 +72,9 @@ def to_html(
|
||||
color = COMMUNITY_COLORS[cid % len(COMMUNITY_COLORS)]
|
||||
net.add_node(
|
||||
node_id,
|
||||
label=data.get("label", node_id),
|
||||
label=sanitize_label(data.get("label", node_id)),
|
||||
color=color,
|
||||
title=(
|
||||
title=sanitize_label(
|
||||
f"Source: {data.get('source_file', 'unknown')}\n"
|
||||
f"Type: {data.get('file_type', 'unknown')}\n"
|
||||
f"Community: {community_labels.get(cid, str(cid)) if community_labels else cid}"
|
||||
|
||||
+33
-33
@@ -70,7 +70,7 @@ def extract_python(path: Path) -> dict:
|
||||
if t == "import_statement":
|
||||
for child in node.children:
|
||||
if child.type in ("dotted_name", "aliased_import"):
|
||||
raw = source[child.start_byte:child.end_byte].decode()
|
||||
raw = source[child.start_byte:child.end_byte].decode("utf-8", errors="replace")
|
||||
module_name = raw.split(" as ")[0].strip().lstrip(".")
|
||||
tgt_nid = _make_id(module_name)
|
||||
add_edge(file_nid, tgt_nid, "imports", node.start_point[0] + 1)
|
||||
@@ -79,7 +79,7 @@ def extract_python(path: Path) -> dict:
|
||||
if t == "import_from_statement":
|
||||
module_node = node.child_by_field_name("module_name")
|
||||
if module_node:
|
||||
raw = source[module_node.start_byte:module_node.end_byte].decode().lstrip(".")
|
||||
raw = source[module_node.start_byte:module_node.end_byte].decode("utf-8", errors="replace").lstrip(".")
|
||||
tgt_nid = _make_id(raw)
|
||||
add_edge(file_nid, tgt_nid, "imports_from", node.start_point[0] + 1)
|
||||
return
|
||||
@@ -88,7 +88,7 @@ def extract_python(path: Path) -> dict:
|
||||
name_node = node.child_by_field_name("name")
|
||||
if not name_node:
|
||||
return
|
||||
class_name = source[name_node.start_byte:name_node.end_byte].decode()
|
||||
class_name = source[name_node.start_byte:name_node.end_byte].decode("utf-8", errors="replace")
|
||||
class_nid = _make_id(stem, class_name)
|
||||
line = node.start_point[0] + 1
|
||||
add_node(class_nid, class_name, line)
|
||||
@@ -99,7 +99,7 @@ def extract_python(path: Path) -> dict:
|
||||
if args:
|
||||
for arg in args.children:
|
||||
if arg.type == "identifier":
|
||||
base = source[arg.start_byte:arg.end_byte].decode()
|
||||
base = source[arg.start_byte:arg.end_byte].decode("utf-8", errors="replace")
|
||||
# Try same-file base first; fall back to a bare stub
|
||||
base_nid = _make_id(stem, base)
|
||||
if base_nid not in seen_ids:
|
||||
@@ -127,7 +127,7 @@ def extract_python(path: Path) -> dict:
|
||||
name_node = node.child_by_field_name("name")
|
||||
if not name_node:
|
||||
return
|
||||
func_name = source[name_node.start_byte:name_node.end_byte].decode()
|
||||
func_name = source[name_node.start_byte:name_node.end_byte].decode("utf-8", errors="replace")
|
||||
line = node.start_point[0] + 1
|
||||
if parent_class_nid:
|
||||
func_nid = _make_id(parent_class_nid, func_name)
|
||||
@@ -170,11 +170,11 @@ def extract_python(path: Path) -> dict:
|
||||
callee_name: str | None = None
|
||||
if func_node:
|
||||
if func_node.type == "identifier":
|
||||
callee_name = source[func_node.start_byte:func_node.end_byte].decode()
|
||||
callee_name = source[func_node.start_byte:func_node.end_byte].decode("utf-8", errors="replace")
|
||||
elif func_node.type == "attribute":
|
||||
attr = func_node.child_by_field_name("attribute")
|
||||
if attr:
|
||||
callee_name = source[attr.start_byte:attr.end_byte].decode()
|
||||
callee_name = source[attr.start_byte:attr.end_byte].decode("utf-8", errors="replace")
|
||||
if callee_name:
|
||||
tgt_nid = label_to_nid.get(callee_name.lower())
|
||||
if tgt_nid and tgt_nid != caller_nid:
|
||||
@@ -273,7 +273,7 @@ def extract_js(path: Path) -> dict:
|
||||
if t == "import_statement":
|
||||
for child in node.children:
|
||||
if child.type == "string":
|
||||
raw = source[child.start_byte:child.end_byte].decode().strip("'\"` ")
|
||||
raw = source[child.start_byte:child.end_byte].decode("utf-8", errors="replace").strip("'\"` ")
|
||||
module_name = raw.lstrip("./").split("/")[-1]
|
||||
if module_name:
|
||||
tgt_nid = _make_id(module_name)
|
||||
@@ -284,7 +284,7 @@ def extract_js(path: Path) -> dict:
|
||||
name_node = node.child_by_field_name("name")
|
||||
if not name_node:
|
||||
return
|
||||
class_name = source[name_node.start_byte:name_node.end_byte].decode()
|
||||
class_name = source[name_node.start_byte:name_node.end_byte].decode("utf-8", errors="replace")
|
||||
class_nid = _make_id(stem, class_name)
|
||||
line = node.start_point[0] + 1
|
||||
add_node(class_nid, class_name, line)
|
||||
@@ -299,7 +299,7 @@ def extract_js(path: Path) -> dict:
|
||||
name_node = node.child_by_field_name("name")
|
||||
if not name_node:
|
||||
return
|
||||
func_name = source[name_node.start_byte:name_node.end_byte].decode()
|
||||
func_name = source[name_node.start_byte:name_node.end_byte].decode("utf-8", errors="replace")
|
||||
line = node.start_point[0] + 1
|
||||
func_nid = _make_id(stem, func_name)
|
||||
add_node(func_nid, f"{func_name}()", line)
|
||||
@@ -313,7 +313,7 @@ def extract_js(path: Path) -> dict:
|
||||
name_node = node.child_by_field_name("name")
|
||||
if not name_node:
|
||||
return
|
||||
method_name = source[name_node.start_byte:name_node.end_byte].decode()
|
||||
method_name = source[name_node.start_byte:name_node.end_byte].decode("utf-8", errors="replace")
|
||||
line = node.start_point[0] + 1
|
||||
method_nid = _make_id(parent_class_nid, method_name)
|
||||
add_node(method_nid, f".{method_name}()", line)
|
||||
@@ -331,7 +331,7 @@ def extract_js(path: Path) -> dict:
|
||||
if value and value.type == "arrow_function":
|
||||
name_node = child.child_by_field_name("name")
|
||||
if name_node:
|
||||
func_name = source[name_node.start_byte:name_node.end_byte].decode()
|
||||
func_name = source[name_node.start_byte:name_node.end_byte].decode("utf-8", errors="replace")
|
||||
line = child.start_point[0] + 1
|
||||
func_nid = _make_id(stem, func_name)
|
||||
add_node(func_nid, f"{func_name}()", line)
|
||||
@@ -362,11 +362,11 @@ def extract_js(path: Path) -> dict:
|
||||
callee_name: str | None = None
|
||||
if func_node:
|
||||
if func_node.type == "identifier":
|
||||
callee_name = source[func_node.start_byte:func_node.end_byte].decode()
|
||||
callee_name = source[func_node.start_byte:func_node.end_byte].decode("utf-8", errors="replace")
|
||||
elif func_node.type == "member_expression":
|
||||
prop = func_node.child_by_field_name("property")
|
||||
if prop:
|
||||
callee_name = source[prop.start_byte:prop.end_byte].decode()
|
||||
callee_name = source[prop.start_byte:prop.end_byte].decode("utf-8", errors="replace")
|
||||
if callee_name:
|
||||
tgt_nid = label_to_nid.get(callee_name.lower())
|
||||
if tgt_nid and tgt_nid != caller_nid:
|
||||
@@ -455,7 +455,7 @@ def extract_go(path: Path) -> dict:
|
||||
if t == "function_declaration":
|
||||
name_node = node.child_by_field_name("name")
|
||||
if name_node:
|
||||
func_name = source[name_node.start_byte:name_node.end_byte].decode()
|
||||
func_name = source[name_node.start_byte:name_node.end_byte].decode("utf-8", errors="replace")
|
||||
line = node.start_point[0] + 1
|
||||
func_nid = _make_id(stem, func_name)
|
||||
add_node(func_nid, f"{func_name}()", line)
|
||||
@@ -473,12 +473,12 @@ def extract_go(path: Path) -> dict:
|
||||
if param.type == "parameter_declaration":
|
||||
type_node = param.child_by_field_name("type")
|
||||
if type_node:
|
||||
raw = source[type_node.start_byte:type_node.end_byte].decode().lstrip("*").strip()
|
||||
raw = source[type_node.start_byte:type_node.end_byte].decode("utf-8", errors="replace").lstrip("*").strip()
|
||||
receiver_type = raw
|
||||
break
|
||||
name_node = node.child_by_field_name("name")
|
||||
if name_node:
|
||||
method_name = source[name_node.start_byte:name_node.end_byte].decode()
|
||||
method_name = source[name_node.start_byte:name_node.end_byte].decode("utf-8", errors="replace")
|
||||
line = node.start_point[0] + 1
|
||||
if receiver_type:
|
||||
parent_nid = _make_id(stem, receiver_type)
|
||||
@@ -500,7 +500,7 @@ def extract_go(path: Path) -> dict:
|
||||
if child.type == "type_spec":
|
||||
name_node = child.child_by_field_name("name")
|
||||
if name_node:
|
||||
type_name = source[name_node.start_byte:name_node.end_byte].decode()
|
||||
type_name = source[name_node.start_byte:name_node.end_byte].decode("utf-8", errors="replace")
|
||||
line = child.start_point[0] + 1
|
||||
type_nid = _make_id(stem, type_name)
|
||||
add_node(type_nid, type_name, line)
|
||||
@@ -514,14 +514,14 @@ def extract_go(path: Path) -> dict:
|
||||
if spec.type == "import_spec":
|
||||
path_node = spec.child_by_field_name("path")
|
||||
if path_node:
|
||||
raw = source[path_node.start_byte:path_node.end_byte].decode().strip('"')
|
||||
raw = source[path_node.start_byte:path_node.end_byte].decode("utf-8", errors="replace").strip('"')
|
||||
module_name = raw.split("/")[-1]
|
||||
tgt_nid = _make_id(module_name)
|
||||
add_edge_raw(file_nid, tgt_nid, "imports_from", spec.start_point[0] + 1)
|
||||
elif child.type == "import_spec":
|
||||
path_node = child.child_by_field_name("path")
|
||||
if path_node:
|
||||
raw = source[path_node.start_byte:path_node.end_byte].decode().strip('"')
|
||||
raw = source[path_node.start_byte:path_node.end_byte].decode("utf-8", errors="replace").strip('"')
|
||||
module_name = raw.split("/")[-1]
|
||||
tgt_nid = _make_id(module_name)
|
||||
add_edge_raw(file_nid, tgt_nid, "imports_from", child.start_point[0] + 1)
|
||||
@@ -548,11 +548,11 @@ def extract_go(path: Path) -> dict:
|
||||
callee_name: str | None = None
|
||||
if func_node:
|
||||
if func_node.type == "identifier":
|
||||
callee_name = source[func_node.start_byte:func_node.end_byte].decode()
|
||||
callee_name = source[func_node.start_byte:func_node.end_byte].decode("utf-8", errors="replace")
|
||||
elif func_node.type == "selector_expression":
|
||||
field = func_node.child_by_field_name("field")
|
||||
if field:
|
||||
callee_name = source[field.start_byte:field.end_byte].decode()
|
||||
callee_name = source[field.start_byte:field.end_byte].decode("utf-8", errors="replace")
|
||||
if callee_name:
|
||||
tgt_nid = label_to_nid.get(callee_name.lower())
|
||||
if tgt_nid and tgt_nid != caller_nid:
|
||||
@@ -641,7 +641,7 @@ def extract_rust(path: Path) -> dict:
|
||||
if t == "function_item":
|
||||
name_node = node.child_by_field_name("name")
|
||||
if name_node:
|
||||
func_name = source[name_node.start_byte:name_node.end_byte].decode()
|
||||
func_name = source[name_node.start_byte:name_node.end_byte].decode("utf-8", errors="replace")
|
||||
line = node.start_point[0] + 1
|
||||
if parent_impl_nid:
|
||||
func_nid = _make_id(parent_impl_nid, func_name)
|
||||
@@ -659,7 +659,7 @@ def extract_rust(path: Path) -> dict:
|
||||
if t in ("struct_item", "enum_item", "trait_item"):
|
||||
name_node = node.child_by_field_name("name")
|
||||
if name_node:
|
||||
item_name = source[name_node.start_byte:name_node.end_byte].decode()
|
||||
item_name = source[name_node.start_byte:name_node.end_byte].decode("utf-8", errors="replace")
|
||||
line = node.start_point[0] + 1
|
||||
item_nid = _make_id(stem, item_name)
|
||||
add_node(item_nid, item_name, line)
|
||||
@@ -670,7 +670,7 @@ def extract_rust(path: Path) -> dict:
|
||||
type_node = node.child_by_field_name("type")
|
||||
impl_nid: str | None = None
|
||||
if type_node:
|
||||
type_name = source[type_node.start_byte:type_node.end_byte].decode().strip()
|
||||
type_name = source[type_node.start_byte:type_node.end_byte].decode("utf-8", errors="replace").strip()
|
||||
impl_nid = _make_id(stem, type_name)
|
||||
add_node(impl_nid, type_name, node.start_point[0] + 1)
|
||||
body = node.child_by_field_name("body")
|
||||
@@ -682,7 +682,7 @@ def extract_rust(path: Path) -> dict:
|
||||
if t == "use_declaration":
|
||||
arg = node.child_by_field_name("argument")
|
||||
if arg:
|
||||
raw = source[arg.start_byte:arg.end_byte].decode()
|
||||
raw = source[arg.start_byte:arg.end_byte].decode("utf-8", errors="replace")
|
||||
clean = raw.split("{")[0].rstrip(":").rstrip("*").rstrip(":")
|
||||
module_name = clean.split("::")[-1].strip()
|
||||
if module_name:
|
||||
@@ -711,15 +711,15 @@ def extract_rust(path: Path) -> dict:
|
||||
callee_name: str | None = None
|
||||
if func_node:
|
||||
if func_node.type == "identifier":
|
||||
callee_name = source[func_node.start_byte:func_node.end_byte].decode()
|
||||
callee_name = source[func_node.start_byte:func_node.end_byte].decode("utf-8", errors="replace")
|
||||
elif func_node.type == "field_expression":
|
||||
field = func_node.child_by_field_name("field")
|
||||
if field:
|
||||
callee_name = source[field.start_byte:field.end_byte].decode()
|
||||
callee_name = source[field.start_byte:field.end_byte].decode("utf-8", errors="replace")
|
||||
elif func_node.type == "scoped_identifier":
|
||||
name = func_node.child_by_field_name("name")
|
||||
if name:
|
||||
callee_name = source[name.start_byte:name.end_byte].decode()
|
||||
callee_name = source[name.start_byte:name.end_byte].decode("utf-8", errors="replace")
|
||||
if callee_name:
|
||||
tgt_nid = label_to_nid.get(callee_name.lower())
|
||||
if tgt_nid and tgt_nid != caller_nid:
|
||||
@@ -830,12 +830,12 @@ def _resolve_cross_file_imports(
|
||||
# Dig into relative_import → dotted_name → identifier
|
||||
for sub in child.children:
|
||||
if sub.type == "dotted_name":
|
||||
raw = source[sub.start_byte:sub.end_byte].decode()
|
||||
raw = source[sub.start_byte:sub.end_byte].decode("utf-8", errors="replace")
|
||||
target_stem = raw.split(".")[-1]
|
||||
break
|
||||
break
|
||||
if child.type == "dotted_name" and target_stem is None:
|
||||
raw = source[child.start_byte:child.end_byte].decode()
|
||||
raw = source[child.start_byte:child.end_byte].decode("utf-8", errors="replace")
|
||||
target_stem = raw.split(".")[-1]
|
||||
|
||||
if not target_stem or target_stem not in stem_to_entities:
|
||||
@@ -853,14 +853,14 @@ def _resolve_cross_file_imports(
|
||||
continue
|
||||
if child.type == "dotted_name":
|
||||
imported_names.append(
|
||||
source[child.start_byte:child.end_byte].decode()
|
||||
source[child.start_byte:child.end_byte].decode("utf-8", errors="replace")
|
||||
)
|
||||
elif child.type == "aliased_import":
|
||||
# `import X as Y` — take the original name
|
||||
name_node = child.child_by_field_name("name")
|
||||
if name_node:
|
||||
imported_names.append(
|
||||
source[name_node.start_byte:name_node.end_byte].decode()
|
||||
source[name_node.start_byte:name_node.end_byte].decode("utf-8", errors="replace")
|
||||
)
|
||||
|
||||
line = node.start_point[0] + 1
|
||||
|
||||
+29
-23
@@ -3,11 +3,13 @@ from __future__ import annotations
|
||||
import json
|
||||
import re
|
||||
import sys
|
||||
import urllib.request
|
||||
import urllib.error
|
||||
import urllib.parse
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
|
||||
from graphify.security import safe_fetch, safe_fetch_text, validate_url
|
||||
|
||||
|
||||
def _safe_filename(url: str, suffix: str) -> str:
|
||||
"""Turn a URL into a safe filename."""
|
||||
@@ -39,9 +41,7 @@ def _detect_url_type(url: str) -> str:
|
||||
|
||||
|
||||
def _fetch_html(url: str) -> str:
|
||||
req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0 graphify/1.0"})
|
||||
with urllib.request.urlopen(req, timeout=15) as resp:
|
||||
return resp.read().decode("utf-8", errors="ignore")
|
||||
return safe_fetch_text(url)
|
||||
|
||||
|
||||
def _html_to_markdown(html: str, url: str) -> str:
|
||||
@@ -175,9 +175,7 @@ def _download_binary(url: str, suffix: str, target_dir: Path) -> Path:
|
||||
"""Download a binary file (PDF, image) directly."""
|
||||
filename = _safe_filename(url, suffix)
|
||||
out_path = target_dir / filename
|
||||
req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0 graphify/1.0"})
|
||||
with urllib.request.urlopen(req, timeout=30) as resp:
|
||||
out_path.write_bytes(resp.read())
|
||||
out_path.write_bytes(safe_fetch(url))
|
||||
return out_path
|
||||
|
||||
|
||||
@@ -190,23 +188,31 @@ def ingest(url: str, target_dir: Path, author: str | None = None, contributor: s
|
||||
target_dir.mkdir(parents=True, exist_ok=True)
|
||||
url_type = _detect_url_type(url)
|
||||
|
||||
if url_type == "pdf":
|
||||
out = _download_binary(url, ".pdf", target_dir)
|
||||
print(f"Downloaded PDF: {out.name}")
|
||||
return out
|
||||
try:
|
||||
validate_url(url)
|
||||
except ValueError as exc:
|
||||
raise ValueError(f"ingest: {exc}") from exc
|
||||
|
||||
if url_type == "image":
|
||||
suffix = Path(urllib.parse.urlparse(url).path).suffix or ".jpg"
|
||||
out = _download_binary(url, suffix, target_dir)
|
||||
print(f"Downloaded image: {out.name}")
|
||||
return out
|
||||
try:
|
||||
if url_type == "pdf":
|
||||
out = _download_binary(url, ".pdf", target_dir)
|
||||
print(f"Downloaded PDF: {out.name}")
|
||||
return out
|
||||
|
||||
if url_type == "tweet":
|
||||
content, filename = _fetch_tweet(url, author, contributor)
|
||||
elif url_type == "arxiv":
|
||||
content, filename = _fetch_arxiv(url, author, contributor)
|
||||
else:
|
||||
content, filename = _fetch_webpage(url, author, contributor)
|
||||
if url_type == "image":
|
||||
suffix = Path(urllib.parse.urlparse(url).path).suffix or ".jpg"
|
||||
out = _download_binary(url, suffix, target_dir)
|
||||
print(f"Downloaded image: {out.name}")
|
||||
return out
|
||||
|
||||
if url_type == "tweet":
|
||||
content, filename = _fetch_tweet(url, author, contributor)
|
||||
elif url_type == "arxiv":
|
||||
content, filename = _fetch_arxiv(url, author, contributor)
|
||||
else:
|
||||
content, filename = _fetch_webpage(url, author, contributor)
|
||||
except (urllib.error.HTTPError, urllib.error.URLError, OSError) as exc:
|
||||
raise RuntimeError(f"ingest: failed to fetch {url!r}: {exc}") from exc
|
||||
|
||||
out_path = target_dir / filename
|
||||
# Avoid overwriting — append counter if needed
|
||||
@@ -245,7 +251,7 @@ def save_query_result(
|
||||
"---",
|
||||
f'type: "{query_type}"',
|
||||
f'date: "{now.isoformat()}"',
|
||||
f'question: "{question.replace(chr(34), chr(39))}"',
|
||||
f'question: "{re.sub(chr(10) + chr(13), " ", question).replace(chr(34), chr(39))}"',
|
||||
'contributor: "graphify"',
|
||||
]
|
||||
if source_nodes:
|
||||
|
||||
@@ -0,0 +1,166 @@
|
||||
# Security helpers — URL validation, safe fetch, path guards, label sanitisation
|
||||
from __future__ import annotations
|
||||
|
||||
import html
|
||||
import re
|
||||
import urllib.error
|
||||
import urllib.parse
|
||||
import urllib.request
|
||||
from pathlib import Path
|
||||
|
||||
_ALLOWED_SCHEMES = {"http", "https"}
|
||||
_MAX_FETCH_BYTES = 52_428_800 # 50 MB hard cap for binary downloads
|
||||
_MAX_TEXT_BYTES = 10_485_760 # 10 MB hard cap for HTML / text
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# URL validation
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def validate_url(url: str) -> str:
|
||||
"""Raise ValueError if *url* is not http or https.
|
||||
|
||||
Blocks file://, ftp://, data:, and any other scheme that could be used
|
||||
for SSRF or local file access.
|
||||
"""
|
||||
parsed = urllib.parse.urlparse(url)
|
||||
if parsed.scheme.lower() not in _ALLOWED_SCHEMES:
|
||||
raise ValueError(
|
||||
f"Blocked URL scheme '{parsed.scheme}' — only http and https are allowed. "
|
||||
f"Got: {url!r}"
|
||||
)
|
||||
return url
|
||||
|
||||
|
||||
class _NoFileRedirectHandler(urllib.request.HTTPRedirectHandler):
|
||||
"""Redirect handler that re-validates every redirect target.
|
||||
|
||||
Prevents open-redirect SSRF attacks where an http:// URL redirects
|
||||
to file:// or an internal address.
|
||||
"""
|
||||
|
||||
def redirect_request(self, req, fp, code, msg, headers, newurl):
|
||||
validate_url(newurl) # raises ValueError if scheme is wrong
|
||||
return super().redirect_request(req, fp, code, msg, headers, newurl)
|
||||
|
||||
|
||||
def _build_opener() -> urllib.request.OpenerDirector:
|
||||
return urllib.request.build_opener(_NoFileRedirectHandler)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Safe fetch
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def safe_fetch(url: str, max_bytes: int = _MAX_FETCH_BYTES, timeout: int = 30) -> bytes:
|
||||
"""Fetch *url* and return raw bytes.
|
||||
|
||||
Protections applied:
|
||||
- URL scheme validated (http / https only)
|
||||
- Redirects re-validated via _NoFileRedirectHandler
|
||||
- Response body capped at *max_bytes* (streaming read)
|
||||
- Non-2xx status raises urllib.error.HTTPError
|
||||
- Network errors propagate as urllib.error.URLError / OSError
|
||||
|
||||
Raises:
|
||||
ValueError — disallowed scheme or redirect target
|
||||
urllib.error.HTTPError — non-2xx HTTP status
|
||||
urllib.error.URLError — DNS / connection failure
|
||||
OSError — size cap exceeded
|
||||
"""
|
||||
validate_url(url)
|
||||
opener = _build_opener()
|
||||
req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0 graphify/1.0"})
|
||||
|
||||
with opener.open(req, timeout=timeout) as resp:
|
||||
# urllib raises HTTPError for non-2xx when using urlopen directly;
|
||||
# with a custom opener we check manually to be safe.
|
||||
status = getattr(resp, "status", None) or getattr(resp, "code", None)
|
||||
if status is not None and not (200 <= status < 300):
|
||||
raise urllib.error.HTTPError(url, status, f"HTTP {status}", {}, None)
|
||||
|
||||
chunks: list[bytes] = []
|
||||
total = 0
|
||||
while True:
|
||||
chunk = resp.read(65_536)
|
||||
if not chunk:
|
||||
break
|
||||
total += len(chunk)
|
||||
if total > max_bytes:
|
||||
raise OSError(
|
||||
f"Response from {url!r} exceeds size limit "
|
||||
f"({max_bytes // 1_048_576} MB). Aborting download."
|
||||
)
|
||||
chunks.append(chunk)
|
||||
|
||||
return b"".join(chunks)
|
||||
|
||||
|
||||
def safe_fetch_text(url: str, max_bytes: int = _MAX_TEXT_BYTES, timeout: int = 15) -> str:
|
||||
"""Fetch *url* and return decoded text (UTF-8, replacing bad bytes).
|
||||
|
||||
Wraps safe_fetch with tighter defaults for HTML / text content.
|
||||
"""
|
||||
raw = safe_fetch(url, max_bytes=max_bytes, timeout=timeout)
|
||||
return raw.decode("utf-8", errors="replace")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Path validation
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def validate_graph_path(path: str | Path, base: Path | None = None) -> Path:
|
||||
"""Resolve *path* and verify it stays inside *base*.
|
||||
|
||||
*base* defaults to the `.graphify` directory relative to CWD.
|
||||
Also requires the base directory to exist, so a caller cannot
|
||||
trick graphify into reading files before any graph has been built.
|
||||
|
||||
Raises:
|
||||
ValueError — path escapes base, or base does not exist
|
||||
FileNotFoundError — resolved path does not exist
|
||||
"""
|
||||
if base is None:
|
||||
base = Path(".graphify").resolve()
|
||||
|
||||
base = base.resolve()
|
||||
if not base.exists():
|
||||
raise ValueError(
|
||||
f"Graph base directory does not exist: {base}. "
|
||||
"Run /graphify first to build the graph."
|
||||
)
|
||||
|
||||
resolved = Path(path).resolve()
|
||||
try:
|
||||
resolved.relative_to(base)
|
||||
except ValueError:
|
||||
raise ValueError(
|
||||
f"Path {path!r} escapes the allowed directory {base}. "
|
||||
"Only paths inside .graphify/ are permitted."
|
||||
)
|
||||
|
||||
if not resolved.exists():
|
||||
raise FileNotFoundError(f"Graph file not found: {resolved}")
|
||||
|
||||
return resolved
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Label sanitisation (mirrors code-review-graph's _sanitize_name pattern)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
_CONTROL_CHAR_RE = re.compile(r"[\x00-\x1f\x7f]")
|
||||
_MAX_LABEL_LEN = 256
|
||||
|
||||
|
||||
def sanitize_label(text: str) -> str:
|
||||
"""Strip control characters, cap length, then HTML-escape.
|
||||
|
||||
Applied to all node labels and edge titles before they are embedded
|
||||
in pyvis HTML output or returned via the MCP server, preventing both
|
||||
XSS and broken visualisations from malformed source identifiers.
|
||||
"""
|
||||
text = _CONTROL_CHAR_RE.sub("", text)
|
||||
if len(text) > _MAX_LABEL_LEN:
|
||||
text = text[:_MAX_LABEL_LEN]
|
||||
return html.escape(text)
|
||||
+13
-4
@@ -5,11 +5,20 @@ import sys
|
||||
from pathlib import Path
|
||||
import networkx as nx
|
||||
from networkx.readwrite import json_graph
|
||||
from graphify.security import validate_graph_path, sanitize_label
|
||||
|
||||
|
||||
def _load_graph(graph_path: str) -> nx.Graph:
|
||||
data = json.loads(Path(graph_path).read_text())
|
||||
return json_graph.node_link_graph(data, edges="links")
|
||||
try:
|
||||
safe = validate_graph_path(graph_path)
|
||||
data = json.loads(safe.read_text())
|
||||
return json_graph.node_link_graph(data, edges="links")
|
||||
except (ValueError, FileNotFoundError) as exc:
|
||||
print(f"error: {exc}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
except json.JSONDecodeError as exc:
|
||||
print(f"error: graph.json is corrupted ({exc}). Re-run /graphify to rebuild.", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
def _communities_from_graph(G: nx.Graph) -> dict[int, list[str]]:
|
||||
@@ -71,12 +80,12 @@ def _subgraph_to_text(G: nx.Graph, nodes: set[str], edges: list[tuple], token_bu
|
||||
lines = []
|
||||
for nid in sorted(nodes, key=lambda n: G.degree(n), reverse=True):
|
||||
d = G.nodes[nid]
|
||||
line = f"NODE {d.get('label', nid)} [src={d.get('source_file', '')} loc={d.get('source_location', '')} community={d.get('community', '')}]"
|
||||
line = f"NODE {sanitize_label(d.get('label', nid))} [src={d.get('source_file', '')} loc={d.get('source_location', '')} community={d.get('community', '')}]"
|
||||
lines.append(line)
|
||||
for u, v in edges:
|
||||
if u in nodes and v in nodes:
|
||||
d = G.edges[u, v]
|
||||
line = f"EDGE {G.nodes[u].get('label', u)} --{d.get('relation', '')} [{d.get('confidence', '')}]--> {G.nodes[v].get('label', v)}"
|
||||
line = f"EDGE {sanitize_label(G.nodes[u].get('label', u))} --{d.get('relation', '')} [{d.get('confidence', '')}]--> {sanitize_label(G.nodes[v].get('label', v))}"
|
||||
lines.append(line)
|
||||
output = "\n".join(lines)
|
||||
if len(output) > char_budget:
|
||||
|
||||
Reference in New Issue
Block a user