fix: corrige 6 bugs mineurs (BUG-035 a BUG-040)
CI / lint (push) Successful in 1m36s
CI / security (push) Successful in 1m4s
CI / test (push) Successful in 3m41s
CI / build (push) Successful in 59s
CI / e2e (push) Successful in 11m8s

This commit is contained in:
2026-09-17 20:05:08 -04:00
parent 2c460022f8
commit 2e2a33cef3
25 changed files with 595 additions and 60 deletions
+27
View File
@@ -743,6 +743,33 @@ class TestSecretRedactor:
result = redact_file_content("hello world this is safe")
assert result == "hello world this is safe"
def test_git_sha_not_redacted(self):
"""BUG-035: a bare git commit SHA must not be mangled."""
from backend.secret_redactor import redact_file_content
sha = "a1b2c3d4e5f60718293a4b5c6d7e8f9012345678"
text = f"commit {sha}\nMerge: {sha}"
assert redact_file_content(text) == text
def test_sha256_checksum_not_redacted(self):
from backend.secret_redactor import redact_file_content
digest = "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"
text = f"sha256:{digest} file.tar.gz"
assert redact_file_content(text) == text
def test_hex_secret_in_context_is_redacted(self):
from backend.secret_redactor import redact_file_content
secret = "0123456789abcdef0123456789abcdef01234567"
result = redact_file_content(f"api_key={secret}")
assert secret not in result
assert "MASQUÉ" in result
def test_ambiguous_bare_hex_left_intact(self):
"""A 40-char hex with no secret/hash keyword stays untouched."""
from backend.secret_redactor import redact_file_content
blob = "deadbeefdeadbeefdeadbeefdeadbeefdeadbeef"
text = f"value {blob} end"
assert redact_file_content(text) == text
# ═══════════════════════════════════════════════════════════════════
# Static / PWA caching (Cloudflare / mobile freshness)
+74
View File
@@ -44,6 +44,22 @@ class TestPasswordHashing:
result = hash_password("ab")
assert result is not None
def test_argon2_memory_recalibrated(self):
"""BUG-038: memory cost must stay at the OWASP 19 MiB recommendation."""
from backend.auth.password import (
ARGON2_MEMORY_COST_KIB,
ARGON2_PARALLELISM,
ARGON2_TIME_COST,
ph,
)
assert ARGON2_MEMORY_COST_KIB == 19456
assert ARGON2_TIME_COST == 2
assert ARGON2_PARALLELISM == 1
assert ph.memory_cost == 19456
assert ph.time_cost == 2
assert ph.parallelism == 1
# ═══════════════════════════════════════════════════════════════════
# JWT Handler
@@ -279,3 +295,61 @@ class TestMiddleware:
assert check_vault_access("Vault1", user) is True
assert check_vault_access("Vault3", user) is False
assert check_vault_access("Vault1", nobody) is False
# ═══════════════════════════════════════════════════════════════════
# Insecure (auth-disabled) deployment guard — BUG-037
# ═══════════════════════════════════════════════════════════════════
class TestInsecureAuthGuard:
def test_is_loopback_host(self):
from backend.auth.middleware import is_loopback_host
assert is_loopback_host(None) is True
assert is_loopback_host("127.0.0.1") is True
assert is_loopback_host("::1") is True
assert is_loopback_host("[::1]") is True
assert is_loopback_host("localhost") is True
assert is_loopback_host("0.0.0.0") is False
assert is_loopback_host("192.168.1.10") is False
def test_bind_host_from_argv(self):
from backend.auth.middleware import bind_host_from_argv
assert bind_host_from_argv(
["uvicorn", "backend.main:app", "--host", "0.0.0.0", "--port", "8080"]
) == "0.0.0.0"
assert bind_host_from_argv(["uvicorn", "app", "--host=127.0.0.1"]) == "127.0.0.1"
assert bind_host_from_argv(["uvicorn", "app"]) is None
def test_guard_refuses_public_bind_without_optin(self, monkeypatch):
from backend import main
monkeypatch.setenv("OBSIGATE_AUTH_ENABLED", "false")
monkeypatch.delenv("OBSIGATE_ALLOW_INSECURE", raising=False)
monkeypatch.setattr("sys.argv", ["uvicorn", "backend.main:app", "--host", "0.0.0.0"])
with pytest.raises(RuntimeError):
main._guard_insecure_auth()
def test_guard_allows_loopback(self, monkeypatch):
from backend import main
monkeypatch.setenv("OBSIGATE_AUTH_ENABLED", "false")
monkeypatch.delenv("OBSIGATE_ALLOW_INSECURE", raising=False)
monkeypatch.setattr("sys.argv", ["uvicorn", "backend.main:app", "--host", "127.0.0.1"])
main._guard_insecure_auth() # must not raise
def test_guard_allows_explicit_optin(self, monkeypatch):
from backend import main
monkeypatch.setenv("OBSIGATE_AUTH_ENABLED", "false")
monkeypatch.setenv("OBSIGATE_ALLOW_INSECURE", "true")
monkeypatch.setattr("sys.argv", ["uvicorn", "backend.main:app", "--host", "0.0.0.0"])
main._guard_insecure_auth() # must not raise
def test_guard_noop_when_auth_enabled(self, monkeypatch):
from backend import main
monkeypatch.setenv("OBSIGATE_AUTH_ENABLED", "true")
monkeypatch.setattr("sys.argv", ["uvicorn", "backend.main:app", "--host", "0.0.0.0"])
main._guard_insecure_auth() # must not raise
+36
View File
@@ -137,6 +137,42 @@ class TestLogin:
})
assert resp.status_code == 401
def test_locked_account_returns_401_not_429(self, auth_client, monkeypatch):
"""BUG-039: a locked account must be indistinguishable from an unknown one."""
import backend.auth.router as auth_router
monkeypatch.setattr(auth_router, "is_locked", lambda username: True)
resp = auth_client.post("/api/auth/login", json={
"username": "admin",
"password": "chab30",
})
assert resp.status_code == 401
assert "verrouill" not in resp.json()["detail"].lower()
def test_account_rate_limited_returns_401(self, auth_client, monkeypatch):
"""BUG-039: per-account throttling must not reveal the account exists."""
import backend.auth.router as auth_router
monkeypatch.setattr(auth_router, "is_account_rate_limited", lambda username: True)
resp = auth_client.post("/api/auth/login", json={
"username": "admin",
"password": "chab30",
})
assert resp.status_code == 401
def test_inactive_account_returns_401(self, auth_client, monkeypatch):
"""BUG-039: a disabled account answers like an unknown user."""
import backend.auth.router as auth_router
monkeypatch.setattr(auth_router, "get_user", lambda username: {
"username": username, "active": False, "password_hash": "x",
})
resp = auth_client.post("/api/auth/login", json={
"username": "admin",
"password": "chab30",
})
assert resp.status_code == 401
def test_login_remember_me(self, auth_client):
resp = auth_client.post("/api/auth/login", json={
"username": "admin",
+51
View File
@@ -73,6 +73,36 @@ def test_authenticate_websocket_invalid_token_returns_none(monkeypatch):
assert authenticate_websocket(_StubWebSocket(cookies={"access_token": "garbage"})) is None
def test_authenticate_websocket_query_token_rejected(monkeypatch):
"""BUG-036: the access token must never be accepted from the query string."""
monkeypatch.setenv("OBSIGATE_AUTH_ENABLED", "true")
from backend.auth.jwt_handler import create_access_token
token = create_access_token({
"username": "u", "role": "user", "vaults": ["*"], "display_name": "U",
})
ws = _StubWebSocket(query={"token": token})
assert authenticate_websocket(ws) is None
def test_authenticate_websocket_cookie_token_accepted(monkeypatch):
"""The HttpOnly access_token cookie remains the supported transport."""
monkeypatch.setenv("OBSIGATE_AUTH_ENABLED", "true")
import backend.auth.user_store as user_store
from backend.auth.jwt_handler import create_access_token
token = create_access_token({
"username": "u", "role": "user", "vaults": ["*"], "display_name": "U",
})
monkeypatch.setattr(user_store, "get_user", lambda username: {
"username": username, "role": "user", "vaults": ["*"],
"display_name": "U", "active": True,
})
user = authenticate_websocket(_StubWebSocket(cookies={"access_token": token}))
assert user is not None
assert user["username"] == "u"
# ---------------------------------------------------------------------------
# Manager unit tests (no WebSocket transport)
# ---------------------------------------------------------------------------
@@ -127,6 +157,27 @@ async def test_on_message_rejects_oversized_update(tmp_path: Path):
assert room.updates == []
@pytest.mark.asyncio
async def test_on_message_rejects_oversized_raw(tmp_path: Path):
"""BUG-036: oversized raw frames are dropped before parsing."""
target = tmp_path / "note.md"
target.write_text("x", encoding="utf-8")
manager = _make_manager(tmp_path)
room = CollabRoom(vault="V", path="note.md", file_path=target)
client = _FakeClient(conn_id=1)
import backend.collab as collab_mod
original = collab_mod.MAX_MESSAGE_CHARS
try:
collab_mod.MAX_MESSAGE_CHARS = 10
await manager._on_message(room, client, json.dumps({"type": "text", "text": "hello"}))
finally:
collab_mod.MAX_MESSAGE_CHARS = original
assert room.pending_text is None
class _FakeWebSocket:
def __init__(self):
self.sent: list[dict] = []
+90 -3
View File
@@ -275,6 +275,7 @@ class TestPdfIndexing:
"""When a vault directory is scanned with .pdf files, they appear in files list.
Uses the public _scan_vault() helper directly — no global state needed.
BUG-040: text extraction is deferred, so the scan only carries metadata.
"""
from backend.indexer import _scan_vault
@@ -286,10 +287,96 @@ class TestPdfIndexing:
names = {f["path"] for f in result["files"]}
assert "a.pdf" in names
assert "b.pdf" in names
# The PDF content should have been extracted.
# The scan defers the expensive text extraction.
a_file = next(f for f in result["files"] if f["path"] == "a.pdf")
assert "ObsiGate test PDF" in (a_file.get("content") or "")
assert "uniqueword0" in (a_file.get("content") or "")
assert a_file["content"] == ""
assert a_file["pdf_text_pending"] is True
class TestPdfLazyEnrichment:
"""BUG-040: PDF text is extracted in a deferred background pass."""
def test_scan_defers_pdf_text_extraction(self, pdf_dir: Path, tmp_path: Path):
from backend.indexer import _scan_vault
vault_root = tmp_path / "vault"
vault_root.mkdir()
shutil.copy2(pdf_dir / "simple.pdf", vault_root / "a.pdf")
result = _scan_vault("v", str(vault_root), {})
a_file = next(f for f in result["files"] if f["path"] == "a.pdf")
assert a_file["content"] == ""
assert a_file["content_preview"] == ""
assert a_file["pdf_text_pending"] is True
@pytest.mark.asyncio
async def test_enrich_pdf_texts_fills_content_and_clears_flag(
self, pdf_dir: Path, tmp_path: Path
):
import backend.indexer as idx
vault_root = tmp_path / "vault"
vault_root.mkdir()
shutil.copy2(pdf_dir / "simple.pdf", vault_root / "a.pdf")
file_info = {
"path": "a.pdf",
"title": "a",
"tags": [],
"content": "",
"content_preview": "",
"size": 0,
"modified": "",
"extension": ".pdf",
"pdf_text_pending": True,
}
with idx._index_lock:
idx.index["LazyV"] = {
"files": [file_info],
"tags": {},
"path": str(vault_root),
"paths": [],
}
try:
count = await idx.enrich_pdf_texts("LazyV")
assert count == 1
assert "ObsiGate test PDF" in file_info["content"]
assert "uniqueword0" in file_info["content"]
assert file_info["content_preview"]
assert "pdf_text_pending" not in file_info
finally:
with idx._index_lock:
idx.index.pop("LazyV", None)
@pytest.mark.asyncio
async def test_enrich_pdf_texts_skips_other_vaults(self, pdf_dir: Path, tmp_path: Path):
import backend.indexer as idx
vault_root = tmp_path / "vault"
vault_root.mkdir()
shutil.copy2(pdf_dir / "simple.pdf", vault_root / "a.pdf")
file_info = {
"path": "a.pdf",
"title": "a",
"tags": [],
"content": "",
"content_preview": "",
"size": 0,
"modified": "",
"extension": ".pdf",
"pdf_text_pending": True,
}
with idx._index_lock:
idx.index["OtherV"] = {
"files": [file_info],
"tags": {},
"path": str(vault_root),
"paths": [],
}
try:
assert await idx.enrich_pdf_texts("LazyV") == 0
assert file_info["content"] == ""
finally:
with idx._index_lock:
idx.index.pop("OtherV", None)
# ── Search filter `ext:` ───────────────────────────────────────────────────