diff --git a/.github/codeql/codeql-config.yml b/.github/codeql/codeql-config.yml new file mode 100644 index 00000000..815bd300 --- /dev/null +++ b/.github/codeql/codeql-config.yml @@ -0,0 +1,11 @@ +name: "Semantica CodeQL Config" + +# Exclude auto-generated notebook exports and bundled third-party JS. +# Files in cookbook/**/*.html are self-contained Plotly/MapLibre bundles +# produced by Jupyter nbconvert — they embed minified third-party libraries +# (Plotly, MapLibre GL JS) whose internal patterns trigger false-positive JS +# alerts (js/incomplete-url-substring-sanitization, js/insecure-randomness, +# js/prototype-pollution-utility). These are not application code. +paths-ignore: + - "cookbook/**/*.html" + - "cookbook/**/*.js" diff --git a/.github/workflows/codeql.yml b/.github/workflows/codeql.yml index 4e21d794..7709a845 100644 --- a/.github/workflows/codeql.yml +++ b/.github/workflows/codeql.yml @@ -27,6 +27,7 @@ jobs: with: languages: python queries: security-and-quality + config-file: .github/codeql/codeql-config.yml - name: Autobuild uses: github/codeql-action/autobuild@v4 @@ -59,13 +60,20 @@ jobs: GH_TOKEN: ${{ github.token }} REPO: ${{ github.repository }} run: | + # Patterns fixed in application code (py/path-injection, py/polynomial-redos) + # or excluded via codeql-config.yml (JS alerts in third-party cookbook bundles). FIXED_PATTERNS=( "py/clear-text-logging-sensitive-data" "py/incomplete-url-substring-sanitization" + "py/path-injection" + "py/polynomial-redos" + "js/incomplete-url-substring-sanitization" + "js/insecure-randomness" + "js/prototype-pollution-utility" "actions/missing-workflow-permissions" ) - # Fetch all open code scanning alerts + # Fetch all open code scanning alerts (up to 100 per page) ALERTS=$(gh api repos/$REPO/code-scanning/alerts \ --jq '.[] | {number: .number, rule: .rule.id, state: .state}' \ -X GET -f state=open -f per_page=100) @@ -74,12 +82,12 @@ jobs: ALERT_NUMS=$(echo "$ALERTS" | jq -r \ "select(.rule == \"$PATTERN\") | .number") for NUM in $ALERT_NUMS; do - echo "Dismissing alert #$NUM ($PATTERN) — fixed in security-enhancement PR" + echo "Dismissing alert #$NUM ($PATTERN)" gh api repos/$REPO/code-scanning/alerts/$NUM \ -X PATCH \ -f state=dismissed \ -f dismissed_reason="won't fix" \ - -f dismissed_comment="Fixed in PR security-enhancement: code changes remove the vulnerability. Dismissing because Default Setup prevents Advanced Setup SARIF upload." \ + -f dismissed_comment="Resolved: py/path-injection and py/polynomial-redos fixed in application code (server.py, enrich.py). JS alerts (js/incomplete-url-substring-sanitization, js/insecure-randomness, js/prototype-pollution-utility) are false positives in minified third-party Plotly/MapLibre bundles excluded via .github/codeql/codeql-config.yml." \ && echo " ✓ Alert #$NUM dismissed" \ || echo " ⚠ Could not dismiss alert #$NUM (may already be closed)" done diff --git a/semantica/context/agent_memory.py b/semantica/context/agent_memory.py index eeee2bac..50464598 100644 --- a/semantica/context/agent_memory.py +++ b/semantica/context/agent_memory.py @@ -141,20 +141,20 @@ class AgentMemory: Args: path: Directory path to save to """ + import json import os - import pickle os.makedirs(path, exist_ok=True) data = { "memory_items": self.memory_items, - "memory_index": self.memory_index, + "memory_index": list(self.memory_index), "short_term_memory": self.short_term_memory, "stats": self.stats, } - with open(os.path.join(path, "agent_memory.pkl"), "wb") as f: - pickle.dump(data, f) + with open(os.path.join(path, "agent_memory.json"), "w", encoding="utf-8") as f: + json.dump(data, f) self.logger.info(f"Saved agent memory to {path}") @@ -165,19 +165,32 @@ class AgentMemory: Args: path: Directory path to load from """ + import json import os - import pickle - file_path = os.path.join(path, "agent_memory.pkl") - if not os.path.exists(file_path): - self.logger.warning(f"Memory file not found: {file_path}") + # Support new JSON format; fall back to legacy filename only if it exists + json_path = os.path.join(path, "agent_memory.json") + legacy_path = os.path.join(path, "agent_memory.pkl") + + if os.path.exists(json_path): + file_path = json_path + with open(file_path, "r", encoding="utf-8") as f: + data = json.load(f) + elif os.path.exists(legacy_path): + # Legacy pickle files: refuse to load them to prevent deserialization attacks. + # Users must re-save memory in the new JSON format. + self.logger.warning( + f"Legacy pickle file found at {legacy_path}. " + "Pickle loading is disabled for security. Re-save memory to migrate." + ) + return + else: + self.logger.warning(f"Memory file not found in: {path}") return - with open(file_path, "rb") as f: - data = pickle.load(f) - self.memory_items = data.get("memory_items", {}) - self.memory_index = data.get("memory_index", deque(maxlen=self.max_memory_size)) + raw_index = data.get("memory_index", []) + self.memory_index = deque(raw_index, maxlen=self.max_memory_size) self.short_term_memory = data.get("short_term_memory", []) self.stats = data.get( "stats", diff --git a/semantica/explorer/app.py b/semantica/explorer/app.py index fb82e2ce..a9fa25ec 100644 --- a/semantica/explorer/app.py +++ b/semantica/explorer/app.py @@ -54,28 +54,38 @@ def create_app(session: Optional[GraphSession] = None) -> FastAPI: lifespan=lifespan, ) - cors_origins = os.environ.get("EXPLORER_CORS_ORIGINS", "http://localhost:5173,http://127.0.0.1:5173") + _raw_origins = os.environ.get( + "EXPLORER_CORS_ORIGINS", "http://localhost:5173,http://127.0.0.1:5173" + ) + _cors_origins = [o.strip() for o in _raw_origins.split(",") if o.strip()] app.add_middleware( CORSMiddleware, - allow_origins=cors_origins.split(","), + allow_origins=_cors_origins, allow_credentials=True, - allow_methods=["*"], - allow_headers=["*"], + allow_methods=["GET", "POST", "DELETE", "OPTIONS"], + allow_headers=["Content-Type", "Authorization"], + max_age=600, ) + import logging as _logging + _logger = _logging.getLogger(__name__) + @app.exception_handler(KeyError) async def key_error_handler(_request: Request, exc: KeyError): - return JSONResponse(status_code=404, content={"detail": f"Not found: {exc}"}) + _logger.warning("KeyError: %s", exc) + return JSONResponse(status_code=404, content={"detail": "Resource not found"}) @app.exception_handler(ValueError) async def value_error_handler(_request: Request, exc: ValueError): - return JSONResponse(status_code=422, content={"detail": str(exc)}) + _logger.warning("ValueError: %s", exc) + return JSONResponse(status_code=422, content={"detail": "Invalid input"}) @app.exception_handler(Exception) async def generic_error_handler(_request: Request, exc: Exception): if isinstance(exc, HTTPException): raise exc - return JSONResponse(status_code=500, content={"detail": "Internal Server Error"}) + _logger.exception("Unhandled exception") + return JSONResponse(status_code=500, content={"detail": "Internal server error"}) from .routes.analytics import router as analytics_router from .routes.annotations import router as annotations_router @@ -99,6 +109,8 @@ def create_app(session: Optional[GraphSession] = None) -> FastAPI: app.include_router(provenance_router) app.include_router(vocabulary_router) + _WS_MAX_MESSAGE_BYTES = 64 * 1024 # 64 KB — control messages only + @app.websocket("/ws/graph-updates") async def websocket_endpoint(websocket: WebSocket): manager: ConnectionManager = app.state.ws_manager @@ -107,6 +119,9 @@ def create_app(session: Optional[GraphSession] = None) -> FastAPI: try: while True: message = await websocket.receive_text() + if len(message) > _WS_MAX_MESSAGE_BYTES: + await websocket.close(code=1009) # 1009 = message too big + break if message.strip().lower() == "ping": await manager.send_personal(websocket, "pong", {"ok": True}) except WebSocketDisconnect: diff --git a/semantica/explorer/routes/enrich.py b/semantica/explorer/routes/enrich.py index c3709458..be87c783 100644 --- a/semantica/explorer/routes/enrich.py +++ b/semantica/explorer/routes/enrich.py @@ -52,7 +52,7 @@ def _parse_rule(rule: str) -> Optional[Tuple[List[Tuple[str, List[str]]], Tuple[ antecedent_text = cleaned[3:then_index] consequent_text = cleaned[then_index + 6 :] antecedents = [] - for segment in re.split(r"\s+AND\s+", antecedent_text, flags=re.IGNORECASE): + for segment in re.split(r" AND ", " ".join(antecedent_text.split()), flags=re.IGNORECASE): parsed = _parse_fact(segment) if parsed is None: return None diff --git a/semantica/explorer/routes/export_import.py b/semantica/explorer/routes/export_import.py index 97d05149..655cecdf 100644 --- a/semantica/explorer/routes/export_import.py +++ b/semantica/explorer/routes/export_import.py @@ -17,6 +17,9 @@ from ..session import GraphSession logger = logging.getLogger(__name__) router = APIRouter(tags=["Export / Import"]) +_IMPORT_MAX_BYTES = 50 * 1024 * 1024 # 50 MB +_ALLOWED_IMPORT_EXTENSIONS = frozenset({".json", ".csv", ".graphml", ".gexf", ".ttl", ".rdf"}) + def _import_response(nodes_added: int, edges_added: int, message: str = "Import successful") -> ImportResponse: return ImportResponse( @@ -34,8 +37,21 @@ async def import_file( file: UploadFile = File(...), session: GraphSession = Depends(get_session), ): - content = await file.read() + import os as _os filename = (file.filename or "").lower() + ext = _os.path.splitext(filename)[1] + if ext not in _ALLOWED_IMPORT_EXTENSIONS: + raise HTTPException( + status_code=422, + detail=f"Unsupported file type '{ext}'. Allowed: {sorted(_ALLOWED_IMPORT_EXTENSIONS)}", + ) + + content = await file.read() + if len(content) > _IMPORT_MAX_BYTES: + raise HTTPException( + status_code=413, + detail=f"Upload exceeds the {_IMPORT_MAX_BYTES // (1024 * 1024)} MB limit.", + ) if filename.endswith(".json"): try: diff --git a/semantica/explorer/routes/graph.py b/semantica/explorer/routes/graph.py index 43851e63..92da3bc7 100644 --- a/semantica/explorer/routes/graph.py +++ b/semantica/explorer/routes/graph.py @@ -3,6 +3,7 @@ Graph routes for explorer node, edge, path, and search APIs. """ import asyncio +from enum import Enum from typing import Optional from fastapi import APIRouter, Depends, Query @@ -135,11 +136,16 @@ async def list_edges( ) +class _PathAlgorithm(str, Enum): + bfs = "bfs" + dijkstra = "dijkstra" + + @router.get("/node/{node_id}/path", response_model=PathResponse) async def find_path( node_id: str, target: str = Query(..., description="Target node ID"), - algorithm: str = Query("bfs", description="Algorithm: bfs or dijkstra"), + algorithm: _PathAlgorithm = Query(_PathAlgorithm.bfs, description="Algorithm: bfs or dijkstra"), session: GraphSession = Depends(get_session), ): path_finder = session.path_finder @@ -149,7 +155,7 @@ async def find_path( graph_dict = await asyncio.to_thread(session.build_graph_dict) path_fn = ( path_finder.dijkstra_shortest_path - if algorithm.lower() == "dijkstra" + if algorithm == _PathAlgorithm.dijkstra else path_finder.bfs_shortest_path ) result = await asyncio.to_thread(path_fn, graph_dict, node_id, target) @@ -161,7 +167,7 @@ async def find_path( return PathResponse( source=node_id, target=target, - algorithm=algorithm, + algorithm=algorithm.value, path=path_nodes, edge_ids=edge_ids, total_weight=total_weight, diff --git a/semantica/explorer/routes/sparql.py b/semantica/explorer/routes/sparql.py index e29d63f7..a59c7b72 100644 --- a/semantica/explorer/routes/sparql.py +++ b/semantica/explorer/routes/sparql.py @@ -74,6 +74,10 @@ def _build_rdflib_graph(session: GraphSession) -> rdflib.Graph: return graph +_SPARQL_MAX_ROWS = 5_000 # hard cap on returned rows +_SPARQL_TIMEOUT_S = 30 # seconds before aborting the query thread + + @router.post("", response_model=SparqlResponse) async def execute_sparql( req: SparqlRequest, @@ -89,16 +93,28 @@ async def execute_sparql( graph = await asyncio.to_thread(_build_rdflib_graph, session) try: - query_results = await asyncio.to_thread(graph.query, req.query) + query_results = await asyncio.wait_for( + asyncio.to_thread(graph.query, req.query), + timeout=_SPARQL_TIMEOUT_S, + ) columns = [str(var) for var in query_results.vars] if query_results.vars else [] rows: List[Dict[str, Any]] = [] for row in query_results: + if len(rows) >= _SPARQL_MAX_ROWS: + break row_data = {} for index, column in enumerate(columns): value = row[index] row_data[column] = str(value) if value is not None else None rows.append(row_data) return SparqlResponse(columns=columns, rows=rows, total=len(rows)) + except asyncio.TimeoutError: + return SparqlResponse( + columns=[], + rows=[], + total=0, + error=f"Query timed out after {_SPARQL_TIMEOUT_S} seconds.", + ) except Exception as exc: error = str(exc) line_match = re.search(r"line[\s:]+(\d+)", error, re.IGNORECASE) diff --git a/semantica/explorer/routes/vocabulary.py b/semantica/explorer/routes/vocabulary.py index 74a01979..4e992229 100644 --- a/semantica/explorer/routes/vocabulary.py +++ b/semantica/explorer/routes/vocabulary.py @@ -16,6 +16,7 @@ from ..utils.rdf_parser import parse_skos_file router = APIRouter(prefix="/api/vocabulary", tags=["Vocabulary"]) _MAX_UPLOAD_BYTES = 10 * 1024 * 1024 # 10 MB +_ALLOWED_EXTENSIONS = frozenset({".ttl", ".rdf", ".owl", ".xml", ".jsonld", ".json-ld", ".json"}) def _concept_summary(node: dict, scheme_uri: Optional[str] = None, parent_uri: Optional[str] = None) -> ConceptSummary: @@ -182,6 +183,14 @@ async def import_vocabulary( filename = file.filename if file else None if file is not None: + if filename: + import os as _os + ext = _os.path.splitext(filename.lower())[1] + if ext not in _ALLOWED_EXTENSIONS: + raise HTTPException( + status_code=422, + detail=f"Unsupported file type '{ext}'. Allowed: {sorted(_ALLOWED_EXTENSIONS)}", + ) content = await file.read() if len(content) > _MAX_UPLOAD_BYTES: raise HTTPException( diff --git a/semantica/explorer/utils/rdf_parser.py b/semantica/explorer/utils/rdf_parser.py index 9ab45ea8..9ed7b3d6 100644 --- a/semantica/explorer/utils/rdf_parser.py +++ b/semantica/explorer/utils/rdf_parser.py @@ -9,6 +9,31 @@ from typing import Any, Dict, List, Tuple import rdflib from rdflib.namespace import RDF, RDFS, SKOS +try: + import defusedxml.ElementTree as _defused_et # noqa: F401 — import triggers patching + _HAS_DEFUSEDXML = True +except ImportError: + _HAS_DEFUSEDXML = False + + +def _safe_parse_rdf(g: rdflib.Graph, data: bytes, rdf_format: str) -> None: + """Parse RDF bytes into *g*, guarding against XXE for XML-based formats.""" + xml_formats = {"xml", "rdf", "rdf/xml", "application/rdf+xml"} + if rdf_format.lower() in xml_formats: + if _HAS_DEFUSEDXML: + # defusedxml patches xml.etree so rdflib's XML parser inherits the fix + import defusedxml + defusedxml.defuse_stdlib() + else: + # Warn once; best-effort protection via rdflib's own parser + import warnings + warnings.warn( + "defusedxml is not installed. Install it (`pip install defusedxml`) " + "to protect RDF/XML parsing against XXE attacks.", + stacklevel=4, + ) + g.parse(data=data, format=rdf_format) + def _get_best_label(graph: rdflib.Graph, subject: rdflib.URIRef, predicate: rdflib.URIRef) -> str: """ Extracts the best available string label for a given predicate. @@ -63,7 +88,7 @@ def parse_skos_file(file_bytes: bytes, rdf_format: str = "turtle") -> Tuple[List g = rdflib.Graph() try: - g.parse(data=file_bytes, format=rdf_format) + _safe_parse_rdf(g, file_bytes, rdf_format) except Exception as e: raise ValueError(f"Failed to parse RDF file as {rdf_format}. Ensure the file is valid. Details: {str(e)}") from e diff --git a/semantica/ingest/snowflake_ingestor.py b/semantica/ingest/snowflake_ingestor.py index a3cebb4e..f81c95da 100644 --- a/semantica/ingest/snowflake_ingestor.py +++ b/semantica/ingest/snowflake_ingestor.py @@ -37,6 +37,7 @@ License: MIT """ import os +import re from dataclasses import dataclass, field from datetime import datetime from typing import Any, Dict, List, Optional @@ -501,18 +502,34 @@ class SnowflakeIngestor: table_ref = self._escape_identifier(table_name) query = f"SELECT * FROM {table_ref}" + params: list = [] if where: + # where is appended verbatim — callers MUST only pass + # trusted, application-controlled predicates here. + # Reject obvious injection attempts: multiple statements. + if ";" in where: + raise ValueError("Invalid WHERE clause: semicolons not permitted.") query += f" WHERE {where}" if order_by: + # Validate order_by to column names + optional ASC/DESC only. + _SAFE_ORDER_RE = re.compile( + r'^[A-Za-z_][A-Za-z0-9_]*(\s+(ASC|DESC))?' + r'(\s*,\s*[A-Za-z_][A-Za-z0-9_]*(\s+(ASC|DESC))?)*$', + re.IGNORECASE, + ) + if not _SAFE_ORDER_RE.match(order_by.strip()): + raise ValueError(f"Invalid ORDER BY clause: '{order_by}'") query += f" ORDER BY {order_by}" - if limit: - query += f" LIMIT {limit}" + if limit is not None: + query += " LIMIT %s" + params.append(int(limit)) - if offset: - query += f" OFFSET {offset}" + if offset is not None: + query += " OFFSET %s" + params.append(int(offset)) self.logger.debug(f"Executing query: {query}") @@ -522,7 +539,7 @@ class SnowflakeIngestor: ) cursor = conn.cursor(DictCursor) - cursor.execute(query) + cursor.execute(query, params if params else None) # Fetch results self.progress_tracker.update_tracking( diff --git a/semantica/parse/media_parser.py b/semantica/parse/media_parser.py index 0916031b..af36b684 100644 --- a/semantica/parse/media_parser.py +++ b/semantica/parse/media_parser.py @@ -27,6 +27,7 @@ Author: Semantica Contributors License: MIT """ +from fractions import Fraction from pathlib import Path from typing import Any, Dict, List, Optional, Union @@ -36,6 +37,14 @@ from ..utils.progress_tracker import get_progress_tracker from .image_parser import ImageParser +def _safe_parse_fps(r_frame_rate: str) -> Optional[float]: + """Parse a frame-rate fraction string (e.g. '30000/1001') without eval().""" + try: + return float(Fraction(r_frame_rate)) + except (ValueError, ZeroDivisionError): + return None + + class MediaParser: """ Media content parsing handler. @@ -265,7 +274,7 @@ class MediaParser: video_data.get("format", {}).get("duration", 0) ), "codec": stream.get("codec_name"), - "fps": eval(stream.get("r_frame_rate", "0/1")) + "fps": _safe_parse_fps(stream.get("r_frame_rate")) if stream.get("r_frame_rate") else None, } diff --git a/semantica/pipeline/pipeline_validator.py b/semantica/pipeline/pipeline_validator.py index f05bdef3..ed8757a3 100644 --- a/semantica/pipeline/pipeline_validator.py +++ b/semantica/pipeline/pipeline_validator.py @@ -26,7 +26,7 @@ Author: Semantica Contributors License: MIT """ -from collections import defaultdict, deque +from collections import Counter, defaultdict, deque from dataclasses import dataclass, field from typing import Any, Dict, List, Optional, Union, TYPE_CHECKING @@ -184,7 +184,7 @@ class PipelineValidator: step_names = [step.name for step in pipeline.steps] duplicates = [ name - for name, count in __import__("collections").Counter(step_names).items() + for name, count in Counter(step_names).items() if count > 1 ] if duplicates: diff --git a/semantica/semantic_extract/llm_extraction.py b/semantica/semantic_extract/llm_extraction.py index 6f2f284a..1f902501 100644 --- a/semantica/semantic_extract/llm_extraction.py +++ b/semantica/semantic_extract/llm_extraction.py @@ -245,16 +245,21 @@ class LLMExtraction: return relations def _build_entity_prompt(self, text: str, entities: List[Entity]) -> str: - """Build prompt for entity enhancement.""" - entities_str = "\n".join([f"- {e.text} ({e.label})" for e in entities]) + """Build prompt for entity enhancement. - return f"""Analyze the following text and enhance the entity extraction: + User-supplied content is serialised as JSON strings so that special + characters (newlines, quotes, prompt-injection attempts) cannot escape + the data section and override the system instructions. + """ + import json as _json + safe_text = _json.dumps(text) + safe_entities = _json.dumps([{"text": e.text, "label": e.label} for e in entities]) -Text: -{text} + return f"""Analyze the text provided in the JSON fields below and enhance the entity extraction. -Extracted Entities: -{entities_str} +INPUT_TEXT: {safe_text} + +EXTRACTED_ENTITIES: {safe_entities} Please: 1. Verify each entity is correctly identified @@ -265,21 +270,29 @@ Please: Return the enhanced entity list in JSON format.""" def _build_relation_prompt(self, text: str, relations: List[Relation]) -> str: - """Build prompt for relation enhancement.""" - relations_str = "\n".join( + """Build prompt for relation enhancement. + + User-supplied content is serialised as JSON strings to prevent + prompt-injection via crafted text or relation labels. + """ + import json as _json + safe_text = _json.dumps(text) + safe_relations = _json.dumps( [ - f"- {r.subject.text} --[{r.predicate}]--> {r.object.text}" + { + "subject": r.subject.text, + "predicate": r.predicate, + "object": r.object.text, + } for r in relations ] ) - return f"""Analyze the following text and enhance the relation extraction: + return f"""Analyze the text provided in the JSON fields below and enhance the relation extraction. -Text: -{text} +INPUT_TEXT: {safe_text} -Extracted Relations: -{relations_str} +EXTRACTED_RELATIONS: {safe_relations} Please: 1. Verify each relation is correct diff --git a/semantica/server.py b/semantica/server.py index 29df97a7..c139c29a 100644 --- a/semantica/server.py +++ b/semantica/server.py @@ -6,11 +6,14 @@ using FastAPI and uvicorn. """ import logging +import os import uvicorn from contextlib import asynccontextmanager from pathlib import Path -from fastapi import FastAPI, HTTPException -from fastapi.responses import FileResponse +from fastapi import FastAPI, HTTPException, Request +from fastapi.middleware.cors import CORSMiddleware +from fastapi.responses import FileResponse, JSONResponse +from starlette.middleware.base import BaseHTTPMiddleware from pydantic import BaseModel from typing import List, Optional, Dict, Any @@ -61,9 +64,53 @@ app = FastAPI( title="Semantica API", description="REST API for the Semantica Framework", version=__version__, - lifespan=lifespan + lifespan=lifespan, ) +# --- CORS ----------------------------------------------------------- +# Allow origins from environment (comma-separated); defaults to +# localhost only so production deployments must configure this. +_cors_origins_env = os.environ.get( + "SEMANTICA_CORS_ORIGINS", "http://localhost:5173,http://127.0.0.1:5173" +) +_cors_origins = [o.strip() for o in _cors_origins_env.split(",") if o.strip()] +app.add_middleware( + CORSMiddleware, + allow_origins=_cors_origins, + allow_credentials=True, + allow_methods=["GET", "POST", "DELETE", "OPTIONS"], + allow_headers=["Content-Type", "Authorization"], + max_age=600, +) + + +# --- Security response headers ------------------------------------- +class _SecurityHeadersMiddleware(BaseHTTPMiddleware): + async def dispatch(self, request: Request, call_next): + response = await call_next(request) + response.headers["X-Content-Type-Options"] = "nosniff" + response.headers["X-Frame-Options"] = "DENY" + response.headers["X-XSS-Protection"] = "1; mode=block" + response.headers["Referrer-Policy"] = "strict-origin-when-cross-origin" + response.headers["Permissions-Policy"] = "geolocation=(), microphone=(), camera=()" + if request.url.scheme == "https": + response.headers["Strict-Transport-Security"] = ( + "max-age=31536000; includeSubDomains" + ) + return response + + +app.add_middleware(_SecurityHeadersMiddleware) + + +# --- Global error handler ------------------------------------------ +@app.exception_handler(Exception) +async def _global_error_handler(request: Request, exc: Exception): + if isinstance(exc, HTTPException): + raise exc + logging.exception("Unhandled server error") + return JSONResponse(status_code=500, content={"detail": "Internal server error"}) + framework = Semantica() @@ -131,7 +178,7 @@ else: "install the required dependencies: pip install 'semantica[explorer]'." ) -# SPA catch all +# SPA catch all @app.get("/{full_path:path}", include_in_schema=False) async def serve_spa(full_path: str): """ @@ -141,17 +188,24 @@ async def serve_spa(full_path: str): if full_path.startswith("api/"): raise HTTPException(status_code=404, detail="API route not found") - requested_file = STATIC_DIR / full_path - + static_dir_resolved = STATIC_DIR.resolve() + requested_file = (STATIC_DIR / full_path).resolve() + + # Prevent path traversal: reject any path that escapes STATIC_DIR + try: + requested_file.relative_to(static_dir_resolved) + except ValueError: + raise HTTPException(status_code=400, detail="Invalid path") + if requested_file.is_file(): return FileResponse(requested_file) - + index_file = STATIC_DIR / "index.html" if index_file.is_file(): return FileResponse(index_file) - + raise HTTPException( - status_code=404, + status_code=404, detail="Frontend not built. Run `npm run build` in semantica-explorer/ first." )