Upload 7 files

src/services/html_sanitizer.py

@@ -0,0 +1,79 @@
+"""Small allow-list sanitizer for answer fragments returned to the browser."""
+
+from html import escape
+from html.parser import HTMLParser
+from urllib.parse import urlparse
+
+
+ALLOWED_TAGS = {"a", "b", "blockquote", "br", "code", "details", "div", "em", "h3", "h4", "hr", "i", "li", "ol", "p", "pre", "span", "strong", "summary", "table", "tbody", "td", "th", "thead", "tr", "ul"}
+VOID_TAGS = {"br", "hr"}
+DROP_CONTENT_TAGS = {"button", "form", "iframe", "input", "math", "object", "script", "style", "svg"}
+GLOBAL_ATTRIBUTES = {"aria-controls", "aria-expanded", "aria-label", "class", "role"}
+TAG_ATTRIBUTES = {"a": {"href", "rel", "target", "title"}, "td": {"colspan", "rowspan"}, "th": {"colspan", "rowspan", "scope"}}
+
+
+def _safe_href(value: str) -> str | None:
+    candidate = value.strip()
+    if not candidate:
+        return None
+    parsed = urlparse(candidate)
+    if parsed.scheme.lower() not in {"https", "mailto", "tel"}:
+        return None
+    return candidate
+
+
+class _Sanitizer(HTMLParser):
+    def __init__(self) -> None:
+        super().__init__(convert_charrefs=True)
+        self.output: list[str] = []
+        self.drop_depth = 0
+
+    def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None:
+        tag = tag.lower()
+        if tag in DROP_CONTENT_TAGS:
+            self.drop_depth += 1
+            return
+        if self.drop_depth or tag not in ALLOWED_TAGS:
+            return
+        rendered: list[str] = []
+        for raw_name, raw_value in attrs:
+            name = raw_name.lower()
+            if name.startswith("on") or name == "style":
+                continue
+            if name not in GLOBAL_ATTRIBUTES and name not in TAG_ATTRIBUTES.get(tag, set()):
+                continue
+            value = raw_value or ""
+            if tag == "a" and name == "href":
+                value = _safe_href(value) or ""
+                if not value:
+                    continue
+            if tag == "a" and name == "target" and value != "_blank":
+                continue
+            rendered.append(f' {name}="{escape(value, quote=True)}"')
+        if tag == "a":
+            rendered = [item for item in rendered if not item.startswith(" rel=")]
+            rendered.append(' rel="noopener noreferrer"')
+        self.output.append(f"<{tag}{''.join(rendered)}>")
+
+    def handle_startendtag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None:
+        self.handle_starttag(tag, attrs)
+
+    def handle_endtag(self, tag: str) -> None:
+        tag = tag.lower()
+        if tag in DROP_CONTENT_TAGS:
+            if self.drop_depth:
+                self.drop_depth -= 1
+            return
+        if not self.drop_depth and tag in ALLOWED_TAGS and tag not in VOID_TAGS:
+            self.output.append(f"</{tag}>")
+
+    def handle_data(self, data: str) -> None:
+        if not self.drop_depth:
+            self.output.append(escape(data, quote=False))
+
+
+def sanitize_answer_html(value: str) -> str:
+    parser = _Sanitizer()
+    parser.feed(value or "")
+    parser.close()
+    return "".join(parser.output)

src/services/http_client.py

@@ -1,44 +1,44 @@
-"""Shared outbound HTTP client factory.
-
-Deliberately plain: the default dual-stack resolver and a moderate timeout.
-
-History: v3.8.2 pinned outbound sockets to IPv4 (local_address="0.0.0.0") to
-chase a suspected IPv6 routing problem. Analytics writes had been succeeding
-before that change and stopped after it, so the pin is reverted. Forcing a
-local address can break egress in containerised hosts, and the original
-diagnosis was not supported by evidence: a 12s GET failed while an 8s POST to
-the same host succeeded, which rules out plain slowness.
-
-Set FORCE_IPV4_LOCAL_ADDRESS to "0.0.0.0" only to test that hypothesis again.
-"""
-import logging
-from typing import Optional
-
-import httpx
-
-logger = logging.getLogger(__name__)
-
-# None = let the OS choose (normal dual-stack behaviour). "0.0.0.0" forces IPv4.
-FORCE_IPV4_LOCAL_ADDRESS: Optional[str] = None
-
-DEFAULT_TIMEOUT = 10.0
-
-
-def make_transport(retries: int = 1) -> httpx.AsyncHTTPTransport:
-    kwargs = {"retries": retries}
-    if FORCE_IPV4_LOCAL_ADDRESS:
-        kwargs["local_address"] = FORCE_IPV4_LOCAL_ADDRESS
-    try:
-        return httpx.AsyncHTTPTransport(**kwargs)
-    except Exception as exc:
-        logger.warning(f"Custom transport unavailable ({exc}); using default")
-        return httpx.AsyncHTTPTransport()
-
-
-def make_client(timeout: float = DEFAULT_TIMEOUT, retries: int = 1) -> httpx.AsyncClient:
-    """AsyncClient with a moderate timeout and one transport-level retry."""
-    return httpx.AsyncClient(
-        timeout=timeout,
-        transport=make_transport(retries=retries),
-        follow_redirects=True,
-    )
+"""Shared outbound HTTP client factory.
+
+Deliberately plain: the default dual-stack resolver and a moderate timeout.
+
+History: v3.8.2 pinned outbound sockets to IPv4 (local_address="0.0.0.0") to
+chase a suspected IPv6 routing problem. Analytics writes had been succeeding
+before that change and stopped after it, so the pin is reverted. Forcing a
+local address can break egress in containerised hosts, and the original
+diagnosis was not supported by evidence: a 12s GET failed while an 8s POST to
+the same host succeeded, which rules out plain slowness.
+
+Set FORCE_IPV4_LOCAL_ADDRESS to "0.0.0.0" only to test that hypothesis again.
+"""
+import logging
+from typing import Optional
+
+import httpx
+
+logger = logging.getLogger(__name__)
+
+# None = let the OS choose (normal dual-stack behaviour). "0.0.0.0" forces IPv4.
+FORCE_IPV4_LOCAL_ADDRESS: Optional[str] = None
+
+DEFAULT_TIMEOUT = 10.0
+
+
+def make_transport(retries: int = 1) -> httpx.AsyncHTTPTransport:
+    kwargs = {"retries": retries}
+    if FORCE_IPV4_LOCAL_ADDRESS:
+        kwargs["local_address"] = FORCE_IPV4_LOCAL_ADDRESS
+    try:
+        return httpx.AsyncHTTPTransport(**kwargs)
+    except Exception as exc:
+        logger.warning(f"Custom transport unavailable ({exc}); using default")
+        return httpx.AsyncHTTPTransport()
+
+
+def make_client(timeout: float = DEFAULT_TIMEOUT, retries: int = 1) -> httpx.AsyncClient:
+    """AsyncClient with a moderate timeout and one transport-level retry."""
+    return httpx.AsyncClient(
+        timeout=timeout,
+        transport=make_transport(retries=retries),
+        follow_redirects=True,
+    )

src/services/rag_service.py

@@ -11,7 +11,8 @@ Fixes applied:
    via a dedicated audit() method to avoid disk reads on every /rag-status poll.
 5. _write / cache saves use os.replace for atomic writes.
 """
-import hashlib
+import hashlib
+import hmac
 import json
 import logging
 import os
@@ -25,8 +26,10 @@ import numpy as np
 from src.config import get_settings
 
 logger = logging.getLogger(__name__)
-CHUNK_MAX_CHARS = 1600
-CHUNK_OVERLAP_CHARS = 200
+CHUNK_WORDS = 260
+CHUNK_OVERLAP_WORDS = 40
+TOKEN_RE = re.compile(r"[a-z0-9][a-z0-9_-]{1,}")
+STOP_WORDS = {"about", "and", "are", "can", "for", "from", "how", "library", "please", "that", "the", "this", "what", "when", "where", "which", "with", "you"}
 
 
 class RAGService:
@@ -59,9 +62,24 @@ class RAGService:
             digest.update(str(stat.st_size).encode("utf-8"))
         return digest.hexdigest()
 
-    def _cache_paths(self) -> tuple[Path, Path, Path]:
+    def _cache_paths(self) -> tuple[Path, Path, Path]:
         cache_dir = self._cache_dir()
-        return cache_dir / "faiss_index", cache_dir / "bm25_cache.json", cache_dir / "kb_state.json"
+        return cache_dir / "faiss_index", cache_dir / "bm25_cache.json", cache_dir / "kb_state.json"
+
+    def _cache_signature(self, faiss_dir: Path, bm25_path: Path) -> str:
+        key = get_settings().origin_token.encode("utf-8")
+        if not key:
+            return ""
+        digest = hmac.new(key, digestmod=hashlib.sha256)
+        for path in sorted([bm25_path, *faiss_dir.glob("*")]):
+            if path.is_file():
+                digest.update(path.name.encode("utf-8"))
+                digest.update(path.read_bytes())
+        return digest.hexdigest()
+
+    @staticmethod
+    def _tokens(value: str) -> List[str]:
+        return TOKEN_RE.findall((value or "").lower())
 
     def _build_reverse_index(self) -> None:
         """Build O(1) content -> index lookup. Call after any corpus change."""
@@ -103,7 +121,7 @@ class RAGService:
             )
             self.bm25_corpus = chunks
             self.bm25_meta = metadatas
-            tokenized = [c.lower().split() for c in chunks]
+            tokenized = [self._tokens(c) for c in chunks]
             self.bm25 = BM25Okapi(tokenized)
             self._build_reverse_index()
             self._save_cached_indexes()
@@ -120,16 +138,20 @@ class RAGService:
             return False
         try:
             state = json.loads(state_path.read_text(encoding="utf-8"))
-            if state.get("kb_hash") != self._kb_hash:
+            if state.get("kb_hash") != self._kb_hash:
                 logger.info("RAGService: KB changed (hash mismatch) — rebuilding indexes")
-                return False
+                return False
+            expected = self._cache_signature(faiss_dir, bm25_path)
+            if not expected or not hmac.compare_digest(str(state.get("signature", "")), expected):
+                logger.warning("RAGService: unsigned or modified cache ignored")
+                return False
             self.vectorstore = FAISS.load_local(
                 str(faiss_dir), self._embeddings, allow_dangerous_deserialization=True
             )
             bm25_cache = json.loads(bm25_path.read_text(encoding="utf-8"))
             self.bm25_corpus = bm25_cache["corpus"]
             self.bm25_meta = bm25_cache["meta"]
-            tokenized = [c.lower().split() for c in self.bm25_corpus]
+            tokenized = [self._tokens(c) for c in self.bm25_corpus]
             self.bm25 = BM25Okapi(tokenized)
             return True
         except Exception as exc:
@@ -155,8 +177,8 @@ class RAGService:
 
         # Atomic write for state
         state_tmp = state_path.with_suffix(".tmp")
-        state_tmp.write_text(
-            json.dumps({"kb_hash": self._kb_hash, "chunk_count": len(self.bm25_corpus)}),
+        state_tmp.write_text(
+            json.dumps({"kb_hash": self._kb_hash, "chunk_count": len(self.bm25_corpus), "signature": self._cache_signature(faiss_dir, bm25_path)}),
             encoding="utf-8",
         )
         os.replace(state_tmp, state_path)
@@ -174,18 +196,25 @@ class RAGService:
                 text = fpath.read_text(encoding="utf-8").strip()
                 if not text:
                     continue
-                source_url = ""
-                title = fpath.stem.replace("_", " ").replace("-", " ")
-                for line in text.splitlines()[:8]:
+                source_url = ""
+                title = fpath.stem.replace("_", " ").replace("-", " ")
+                authority = "official"
+                reviewed = ""
+                for line in text.splitlines()[:8]:
                     line = line.strip()
                     if line.startswith("SOURCE:"):
                         source_url = line.replace("SOURCE:", "").strip()
-                    elif line.startswith("TITLE:"):
-                        title = line.replace("TITLE:", "").strip()
+                    elif line.startswith("TITLE:"):
+                        title = line.replace("TITLE:", "").strip()
+                    elif line.startswith("AUTHORITY:"):
+                        authority = line.replace("AUTHORITY:", "").strip()[:40]
+                    elif line.startswith("LAST_REVIEWED:"):
+                        reviewed = line.replace("LAST_REVIEWED:", "").strip()[:20]
                 for chunk in self._chunk_text(text):
                     all_chunks.append(chunk)
                     all_meta.append(
-                        {"source": source_url, "title": title, "filename": fpath.name}
+                        {"source": source_url, "title": title, "filename": fpath.name,
+                         "authority": authority, "last_reviewed": reviewed}
                     )
             except Exception as exc:
                 logger.warning("Failed to load %s: %s", fpath.name, exc)
@@ -199,18 +228,17 @@ class RAGService:
             section = section.strip()
             if not section or len(section) < 30:
                 continue
-            if len(section) <= CHUNK_MAX_CHARS:
-                chunks.append(section)
-            else:
-                start = 0
-                while start < len(section):
-                    end = start + CHUNK_MAX_CHARS
-                    chunk = section[start:end].strip()
-                    if chunk:
-                        chunks.append(chunk)
-                    if end >= len(section):
-                        break
-                    start += CHUNK_MAX_CHARS - CHUNK_OVERLAP_CHARS
+            words = section.split()
+            if len(words) <= CHUNK_WORDS:
+                chunks.append(section)
+                continue
+            step = CHUNK_WORDS - CHUNK_OVERLAP_WORDS
+            for start in range(0, len(words), step):
+                chunk = " ".join(words[start:start + CHUNK_WORDS]).strip()
+                if chunk:
+                    chunks.append(chunk)
+                if start + CHUNK_WORDS >= len(words):
+                    break
         return chunks
 
     async def hybrid_search(self, query: str, top_k: int = 5, alpha: float = 0.6) -> List[dict]:
@@ -224,7 +252,7 @@ class RAGService:
             dense_results = await self.vectorstore.asimilarity_search_with_score(
                 query, k=fetch_k
             )
-            tokenized_query = query.lower().split()
+            tokenized_query = self._tokens(query)
             bm25_scores = self.bm25.get_scores(tokenized_query)
             top_bm25_idx = np.argsort(bm25_scores)[::-1][:fetch_k].tolist()
 
@@ -240,17 +268,25 @@ class RAGService:
                 rrf[idx] = rrf.get(idx, 0.0) + (1 - alpha) * (1.0 / (K + rank + 1))
 
             sorted_idx = sorted(rrf, key=lambda i: rrf[i], reverse=True)[:top_k]
-            results = []
-            for idx in sorted_idx:
-                if idx < len(self.bm25_corpus):
-                    meta = self.bm25_meta[idx]
-                    results.append(
+            results = []
+            meaningful_query = {token for token in tokenized_query if token not in STOP_WORDS and len(token) > 2}
+            for idx in sorted_idx:
+                if idx < len(self.bm25_corpus):
+                    meta = self.bm25_meta[idx]
+                    chunk_tokens = set(self._tokens(self.bm25_corpus[idx]))
+                    lexical_overlap = len(meaningful_query & chunk_tokens)
+                    if meaningful_query and lexical_overlap == 0:
+                        continue
+                    results.append(
                         {
                             "content": self.bm25_corpus[idx],
                             "source": meta.get("source", ""),
                             "title": meta.get("title", ""),
                             "filename": meta.get("filename", ""),
-                            "score": round(rrf[idx], 6),
+                            "score": round(rrf[idx], 6),
+                            "lexical_overlap": lexical_overlap,
+                            "authority": meta.get("authority", "official"),
+                            "last_reviewed": meta.get("last_reviewed", ""),
                         }
                     )
             return results