link guard www + cleanup

app/links.py

@@ -20,36 +20,47 @@ from sqlalchemy.ext.asyncio import AsyncSession
 from app.models import KnowledgeSource
 
 _MD_LINK = re.compile(r"\[([^\]]+)\]\((https?://[^\s)]+)\)")
-_BARE_URL = re.compile(r"https?://[^\s<>\"')]+")
+_BARE_URL = re.compile(r"https?://[^\s<>\"')*\]]+")
+
+
+def _norm_host(host: str) -> str:
+    return host.lower().removeprefix("www.")
 
 
 async def allowed_links(db: AsyncSession, tenant_id: int | None) -> tuple[set[str], set[str]]:
-    """Return (exact_urls, hosts) the bot is allowed to cite for this tenant."""
+    """Return (exact_urls, hosts) the bot is allowed to cite for this tenant.
+    URLs and hosts are normalized (no trailing slash, www-insensitive)."""
     rows = (
         await db.execute(
             select(KnowledgeSource).where(KnowledgeSource.tenant_id == tenant_id)
         )
     ).scalars().all()
-    exact: set[str] = set()
+    raw: set[str] = set()
     for s in rows:
         for u in (s.urls or []):
-            exact.add(u.rstrip("/"))
+            raw.add(u)
         if (s.location or "").startswith(("http://", "https://")):
-            exact.add(s.location.rstrip("/"))
-    hosts = {urlparse(u).netloc.lower() for u in exact if urlparse(u).netloc}
+            raw.add(s.location)
+    exact = {_canon(u) for u in raw}
+    hosts = {_norm_host(urlparse(u).netloc) for u in raw if urlparse(u).netloc}
     return exact, hosts
 
 
+def _canon(url: str) -> str:
+    """Scheme-less, www-less, slash-less key for comparing URLs."""
+    p = urlparse(url.rstrip(".,);:*]’'\"/"))
+    return f"{_norm_host(p.netloc)}{p.path}".rstrip("/").lower()
+
+
 def _fix_url(url: str, exact: set[str], hosts: set[str]) -> str | None:
     """Return the URL to use, or None if it should be dropped entirely."""
     clean = url.rstrip(".,);:*]’'\"")
-    if clean.rstrip("/") in exact:
-        return clean
-    parsed = urlparse(clean)
-    host = parsed.netloc.lower()
+    if _canon(clean) in exact:
+        return clean  # a verified real URL — keep exactly as written
+    host = _norm_host(urlparse(clean).netloc)
     if host in hosts:
         # known store domain but an unverified path → send to the real homepage
-        return f"{parsed.scheme}://{host}"
+        return f"https://{host}"
     return None  # unknown/foreign domain → drop
 
 
@@ -69,5 +80,9 @@ def sanitize_links(text: str, exact: set[str], hosts: set[str]) -> str:
         return fixed if fixed else ""
 
     out = _BARE_URL.sub(_bare, out)
+    # tidy up artefacts left when a link was removed (empty bold/brackets, gaps)
+    out = re.sub(r"\*\*\s*\*\*", "", out)
+    out = re.sub(r"\[\s*\]", "", out)
     out = re.sub(r"[ \t]{2,}", " ", out)
-    return re.sub(r" ([,.;:!?])", r"\1", out).strip()
+    out = re.sub(r" ([,.;:!?])", r"\1", out)
+    return out.strip()

tests/test_links.py

@@ -3,22 +3,26 @@ from __future__ import annotations
 from app.links import allowed_links, sanitize_links
 from app.models import KnowledgeSource
 
+# exact is the CANONICAL form produced by allowed_links (www/scheme/slash-less)
+EXACT = {"flexigobe.com", "flexigobe.com/catalogo.pdf"}
+HOSTS = {"flexigobe.com"}
+
 
 def test_sanitize_keeps_real_drops_fake_and_fixes_path():
-    exact = {"https://flexigobe.com", "https://flexigobe.com/catalogo.pdf"}
-    hosts = {"flexigobe.com"}
-    # real allowlisted link kept
+    # real allowlisted link kept exactly
     assert "https://flexigobe.com/catalogo.pdf" in sanitize_links(
-        "Catálogo: https://flexigobe.com/catalogo.pdf", exact, hosts
+        "Catálogo: https://flexigobe.com/catalogo.pdf", EXACT, HOSTS
     )
     # invented path on the known store domain → downgraded to the real homepage
-    out = sanitize_links("Más info: https://flexigobe.com/about", exact, hosts)
+    out = sanitize_links("Más info: https://flexigobe.com/about", EXACT, HOSTS)
     assert "/about" not in out and "https://flexigobe.com" in out
+    # www variant of a known host is recognized (not treated as foreign)
+    out_www = sanitize_links("Web: https://www.flexigobe.com/inventado", EXACT, HOSTS)
+    assert "flexigobe.com" in out_www and "/inventado" not in out_www
     # foreign/unknown domain → removed
-    out2 = sanitize_links("Visita https://evil.example.com/x ahora", exact, hosts)
-    assert "evil.example.com" not in out2
+    assert "evil.example.com" not in sanitize_links("Visita https://evil.example.com/x", EXACT, HOSTS)
     # markdown link with invented path → label kept, link fixed
-    out3 = sanitize_links("Lee [Sobre nosotros](https://flexigobe.com/about) aquí", exact, hosts)
+    out3 = sanitize_links("Lee [Sobre nosotros](https://flexigobe.com/about) aquí", EXACT, HOSTS)
     assert "Sobre nosotros" in out3 and "/about" not in out3
 
 
@@ -35,7 +39,7 @@ async def test_allowed_links_built_from_source_urls(db_session):
     )
     await db_session.flush()
     exact, hosts = await allowed_links(db_session, 7)
-    assert "https://shop.example/faq" in exact
+    assert "shop.example/faq" in exact  # canonical form
     assert "shop.example" in hosts
     # another tenant's links must not leak in
     exact2, _ = await allowed_links(db_session, 999)