From 84117abd13d788e6e9fe75b2c6d8209c328e4a80 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sun, 29 Mar 2026 17:08:39 +0000 Subject: [PATCH] fix: resolve CodeQL alerts, improve test coverage, fix StaticPool for HTTP tests Agent-Logs-Url: https://github.com/christianlouis/dmarq/sessions/07b58025-2aeb-4e81-a168-7a6fccdc3569 Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com> --- .../app/api/api_v1/endpoints/mail_sources.py | 7 +- backend/app/services/imap_client.py | 4 +- backend/app/tests/conftest.py | 42 +- backend/app/tests/test_mail_sources.py | 371 +++++++++++++++++- 4 files changed, 405 insertions(+), 19 deletions(-) diff --git a/backend/app/api/api_v1/endpoints/mail_sources.py b/backend/app/api/api_v1/endpoints/mail_sources.py index da36f0f..5b7ddf8 100644 --- a/backend/app/api/api_v1/endpoints/mail_sources.py +++ b/backend/app/api/api_v1/endpoints/mail_sources.py @@ -92,6 +92,11 @@ class TestConnectionRequest(BaseModel): # --------------------------------------------------------------------------- +def _sanitize_for_log(value: object) -> str: + """Remove CR/LF characters from a value to prevent log injection attacks.""" + return str(value).replace("\r", "").replace("\n", " ") + + def _get_source_or_404(source_id: int, db: Session) -> MailSource: source = db.query(MailSource).filter(MailSource.id == source_id).first() if source is None: @@ -210,7 +215,7 @@ async def delete_mail_source( source = _get_source_or_404(source_id, db) db.delete(source) db.commit() - logger.info("Deleted mail source id=%d", source_id) + logger.info("Deleted mail source id=%s", _sanitize_for_log(source_id)) @router.post("/{source_id}/toggle", response_model=MailSourceResponse) diff --git a/backend/app/services/imap_client.py b/backend/app/services/imap_client.py index fd73b3b..2eb4b98 100644 --- a/backend/app/services/imap_client.py +++ b/backend/app/services/imap_client.py @@ -132,7 +132,7 @@ class IMAPClient: return True, "Connection successful", stats except Exception as e: # pylint: disable=broad-exception-caught logger.error("IMAP connection test failed: %s", str(e)) - return False, f"Connection failed: {str(e)}", {} + return False, "Connection failed. Check server address and credentials.", {} def _process_single_email(self, mail, email_id: bytes, stats: dict) -> None: """Fetch, parse, and store DMARC attachments from one email message.""" @@ -229,7 +229,7 @@ class IMAPClient: logger.error("Error fetching DMARC reports: %s", str(e)) return { "success": False, - "error": f"Error connecting to mailbox: {str(e)}", + "error": "Error connecting to mailbox. Check server logs for details.", "processed": 0, } diff --git a/backend/app/tests/conftest.py b/backend/app/tests/conftest.py index fb7887d..9d238af 100644 --- a/backend/app/tests/conftest.py +++ b/backend/app/tests/conftest.py @@ -4,12 +4,14 @@ from fastapi import FastAPI from fastapi.testclient import TestClient from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker +from sqlalchemy.pool import StaticPool import app.models.domain # noqa: F401 # pylint: disable=unused-import -import app.models.mail_source # noqa: F401 # pylint: disable=unused-import +import app.models.mail_source as _mail_source_model # noqa: F401 # pylint: disable=unused-import import app.models.report # noqa: F401 # pylint: disable=unused-import import app.models.user # noqa: F401 # pylint: disable=unused-import from app.core.database import Base, get_db +from app.core.security import require_admin_auth from app.main import create_app from app.services.report_store import ReportStore @@ -23,8 +25,17 @@ def test_app() -> FastAPI: @pytest.fixture() def db_session(): - """Create a fresh in-memory SQLite database session per test.""" - engine = create_engine("sqlite://", connect_args={"check_same_thread": False}) + """Create a fresh in-memory SQLite database session per test. + + ``StaticPool`` ensures every SQLAlchemy operation reuses the same + underlying DBAPI connection so the in-memory database (and its tables) + persist for the full duration of the test, even across commits. + """ + engine = create_engine( + "sqlite://", + connect_args={"check_same_thread": False}, + poolclass=StaticPool, + ) Base.metadata.create_all(engine) TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) db = TestingSessionLocal() @@ -59,3 +70,28 @@ def _reset_report_store(): store.clear() yield store.clear() + + +@pytest.fixture() +def authed_client(test_app: FastAPI, db_session): # pylint: disable=redefined-outer-name + """ + TestClient with both DB and admin-auth dependency overrides. + + Bypasses ``require_admin_auth`` so tests can call admin-only endpoints + without needing a real API key or JWT token. + """ + + async def mock_admin_auth(): + return {"auth_type": "api_key", "api_key": "test-key"} + + def override_get_db(): + try: + yield db_session + finally: + pass + + test_app.dependency_overrides[get_db] = override_get_db + test_app.dependency_overrides[require_admin_auth] = mock_admin_auth + with TestClient(test_app) as test_client: + yield test_client + test_app.dependency_overrides.clear() diff --git a/backend/app/tests/test_mail_sources.py b/backend/app/tests/test_mail_sources.py index 0814670..1ad9488 100644 --- a/backend/app/tests/test_mail_sources.py +++ b/backend/app/tests/test_mail_sources.py @@ -2,6 +2,9 @@ Tests for MailSource model and mail-sources API endpoints. """ +from unittest.mock import MagicMock, patch + +import pytest from fastapi.testclient import TestClient from sqlalchemy.orm import Session @@ -71,24 +74,19 @@ class TestMailSourceModel: class TestMailSourcesAPI: - """Integration tests for /api/v1/mail-sources endpoints.""" + """Integration tests for /api/v1/mail-sources endpoints (no auth).""" - API_KEY_HEADER = {"X-API-Key": "test-key"} # will be set in fixture - - def _get_headers(self, client: TestClient) -> dict: - """Return auth headers using the running app's in-memory API key.""" - # The test app generates its own key in-memory; we can't predict it. - # Use the TestClient without auth to check 401, and skip auth-required tests - # by injecting the dependency override instead. - return {} - - def test_list_empty(self, client: TestClient): + def test_list_requires_auth(self, client: TestClient): resp = client.get("/api/v1/mail-sources") # Without auth, expect 401 or 403 assert resp.status_code in (401, 403) - def test_create_and_list(self, client: TestClient, db_session: Session): - """Create a mail source directly in DB and list via authenticated-less read.""" + def test_create_requires_auth(self, client: TestClient): + resp = client.post("/api/v1/mail-sources", json={"name": "x", "method": "IMAP"}) + assert resp.status_code in (401, 403) + + def test_model_create_and_list(self, client: TestClient, db_session: Session): + """Create a mail source directly in DB and verify it's retrievable.""" source = MailSource( name="Direct DB Source", method="IMAP", @@ -153,3 +151,350 @@ class TestMailSourcesAPI: assert "Enabled A" in names assert "Enabled B" in names assert "Disabled" not in names + + +# --------------------------------------------------------------------------- +# Authenticated HTTP API tests (uses authed_client fixture from conftest) +# --------------------------------------------------------------------------- + + +class TestMailSourcesAPIAuthed: + """HTTP-level tests using the authed_client fixture (auth dependency bypassed).""" + + # ------------------------------------------------------------------ + # List + # ------------------------------------------------------------------ + + def test_list_empty(self, authed_client: TestClient): + resp = authed_client.get("/api/v1/mail-sources") + assert resp.status_code == 200 + assert resp.json() == [] + + # ------------------------------------------------------------------ + # Create + # ------------------------------------------------------------------ + + def test_create_imap_source(self, authed_client: TestClient): + payload = { + "name": "My IMAP", + "method": "IMAP", + "server": "imap.example.com", + "port": 993, + "username": "user@example.com", + "password": "s3cr3t", + "use_ssl": True, + "folder": "INBOX", + "polling_interval": 60, + "enabled": True, + } + resp = authed_client.post("/api/v1/mail-sources", json=payload) + assert resp.status_code == 201 + data = resp.json() + assert data["name"] == "My IMAP" + assert data["method"] == "IMAP" + assert data["server"] == "imap.example.com" + assert data["password"] == "**redacted**" + assert data["id"] is not None + + def test_create_normalizes_method_to_uppercase(self, authed_client: TestClient): + payload = {"name": "lowercase method", "method": "imap"} + resp = authed_client.post("/api/v1/mail-sources", json=payload) + assert resp.status_code == 201 + assert resp.json()["method"] == "IMAP" + + # ------------------------------------------------------------------ + # Get single + # ------------------------------------------------------------------ + + def test_get_existing_source(self, authed_client: TestClient): + create_resp = authed_client.post( + "/api/v1/mail-sources", json={"name": "Get Test", "method": "IMAP"} + ) + source_id = create_resp.json()["id"] + + resp = authed_client.get(f"/api/v1/mail-sources/{source_id}") + assert resp.status_code == 200 + assert resp.json()["id"] == source_id + assert resp.json()["name"] == "Get Test" + + def test_get_nonexistent_source_returns_404(self, authed_client: TestClient): + resp = authed_client.get("/api/v1/mail-sources/99999") + assert resp.status_code == 404 + + # ------------------------------------------------------------------ + # Update + # ------------------------------------------------------------------ + + def test_update_name(self, authed_client: TestClient): + create_resp = authed_client.post( + "/api/v1/mail-sources", json={"name": "Original", "method": "IMAP"} + ) + source_id = create_resp.json()["id"] + + update_resp = authed_client.put( + f"/api/v1/mail-sources/{source_id}", json={"name": "Updated"} + ) + assert update_resp.status_code == 200 + assert update_resp.json()["name"] == "Updated" + + def test_update_method_normalizes_uppercase(self, authed_client: TestClient): + create_resp = authed_client.post( + "/api/v1/mail-sources", json={"name": "MethodTest", "method": "IMAP"} + ) + source_id = create_resp.json()["id"] + + update_resp = authed_client.put( + f"/api/v1/mail-sources/{source_id}", json={"method": "pop3"} + ) + assert update_resp.status_code == 200 + assert update_resp.json()["method"] == "POP3" + + def test_update_nonexistent_source_returns_404(self, authed_client: TestClient): + resp = authed_client.put("/api/v1/mail-sources/99999", json={"name": "x"}) + assert resp.status_code == 404 + + # ------------------------------------------------------------------ + # Delete + # ------------------------------------------------------------------ + + def test_delete_source(self, authed_client: TestClient): + create_resp = authed_client.post( + "/api/v1/mail-sources", json={"name": "Delete Me", "method": "IMAP"} + ) + source_id = create_resp.json()["id"] + + del_resp = authed_client.delete(f"/api/v1/mail-sources/{source_id}") + assert del_resp.status_code == 204 + + # Verify gone + get_resp = authed_client.get(f"/api/v1/mail-sources/{source_id}") + assert get_resp.status_code == 404 + + def test_delete_nonexistent_source_returns_404(self, authed_client: TestClient): + resp = authed_client.delete("/api/v1/mail-sources/99999") + assert resp.status_code == 404 + + # ------------------------------------------------------------------ + # List after creates + # ------------------------------------------------------------------ + + def test_list_multiple_sources(self, authed_client: TestClient): + for i in range(3): + authed_client.post( + "/api/v1/mail-sources", json={"name": f"Source {i}", "method": "IMAP"} + ) + resp = authed_client.get("/api/v1/mail-sources") + assert resp.status_code == 200 + assert len(resp.json()) == 3 + + # ------------------------------------------------------------------ + # Toggle + # ------------------------------------------------------------------ + + def test_toggle_disables_then_enables(self, authed_client: TestClient): + create_resp = authed_client.post( + "/api/v1/mail-sources", json={"name": "Toggle", "method": "IMAP", "enabled": True} + ) + source_id = create_resp.json()["id"] + + toggle_resp = authed_client.post(f"/api/v1/mail-sources/{source_id}/toggle") + assert toggle_resp.status_code == 200 + assert toggle_resp.json()["enabled"] is False + + toggle_resp2 = authed_client.post(f"/api/v1/mail-sources/{source_id}/toggle") + assert toggle_resp2.status_code == 200 + assert toggle_resp2.json()["enabled"] is True + + def test_toggle_nonexistent_returns_404(self, authed_client: TestClient): + resp = authed_client.post("/api/v1/mail-sources/99999/toggle") + assert resp.status_code == 404 + + # ------------------------------------------------------------------ + # Test stored source + # ------------------------------------------------------------------ + + def test_test_stored_imap_source_success(self, authed_client: TestClient): + create_resp = authed_client.post( + "/api/v1/mail-sources", + json={ + "name": "IMAP Test", + "method": "IMAP", + "server": "imap.example.com", + "username": "u", + "password": "p", + }, + ) + source_id = create_resp.json()["id"] + + mock_stats = { + "message_count": 10, + "unread_count": 2, + "dmarc_count": 1, + "available_mailboxes": ["INBOX"], + } + mock_client = MagicMock() + mock_client.test_connection.return_value = (True, "Connection successful", mock_stats) + + with patch("app.api.api_v1.endpoints.mail_sources.IMAPClient", return_value=mock_client): + resp = authed_client.post(f"/api/v1/mail-sources/{source_id}/test") + + assert resp.status_code == 200 + data = resp.json() + assert data["success"] is True + assert data["message"] == "Connection successful" + assert data["message_count"] == 10 + + def test_test_stored_imap_source_failure(self, authed_client: TestClient): + create_resp = authed_client.post( + "/api/v1/mail-sources", + json={"name": "IMAP Fail", "method": "IMAP", "server": "bad.host"}, + ) + source_id = create_resp.json()["id"] + + mock_client = MagicMock() + mock_client.test_connection.return_value = ( + False, + "Connection failed. Check server address and credentials.", + {}, + ) + + with patch("app.api.api_v1.endpoints.mail_sources.IMAPClient", return_value=mock_client): + resp = authed_client.post(f"/api/v1/mail-sources/{source_id}/test") + + assert resp.status_code == 200 + data = resp.json() + assert data["success"] is False + + def test_test_stored_non_imap_source(self, authed_client: TestClient): + create_resp = authed_client.post( + "/api/v1/mail-sources", json={"name": "POP3 Source", "method": "POP3"} + ) + source_id = create_resp.json()["id"] + + resp = authed_client.post(f"/api/v1/mail-sources/{source_id}/test") + assert resp.status_code == 200 + data = resp.json() + assert data["success"] is False + assert "not yet implemented" in data["message"] + + def test_test_stored_nonexistent_returns_404(self, authed_client: TestClient): + resp = authed_client.post("/api/v1/mail-sources/99999/test") + assert resp.status_code == 404 + + # ------------------------------------------------------------------ + # Ad-hoc test connection + # ------------------------------------------------------------------ + + def test_adhoc_imap_success(self, authed_client: TestClient): + payload = { + "server": "imap.example.com", + "port": 993, + "username": "user@example.com", + "password": "secret", + "ssl": True, + "method": "IMAP", + } + mock_stats = { + "message_count": 5, + "unread_count": 1, + "dmarc_count": 0, + "available_mailboxes": ["INBOX"], + } + mock_client = MagicMock() + mock_client.test_connection.return_value = (True, "Connection successful", mock_stats) + + with patch("app.api.api_v1.endpoints.mail_sources.IMAPClient", return_value=mock_client): + resp = authed_client.post("/api/v1/mail-sources/test-connection", json=payload) + + assert resp.status_code == 200 + data = resp.json() + assert data["success"] is True + assert data["message_count"] == 5 + + def test_adhoc_non_imap_returns_not_implemented(self, authed_client: TestClient): + payload = { + "server": "pop3.example.com", + "port": 110, + "username": "u", + "password": "p", + "ssl": False, + "method": "POP3", + } + resp = authed_client.post("/api/v1/mail-sources/test-connection", json=payload) + assert resp.status_code == 200 + data = resp.json() + assert data["success"] is False + assert "not yet implemented" in data["message"] + + def test_adhoc_gmail_api_returns_not_implemented(self, authed_client: TestClient): + payload = {"method": "GMAIL_API"} + resp = authed_client.post("/api/v1/mail-sources/test-connection", json=payload) + assert resp.status_code == 200 + assert resp.json()["success"] is False + assert "not yet implemented" in resp.json()["message"] + + +# --------------------------------------------------------------------------- +# _sanitize_for_log helper +# --------------------------------------------------------------------------- + + +class TestSanitizeForLog: + """Unit tests for the _sanitize_for_log helper.""" + + def test_strips_newline(self): + from app.api.api_v1.endpoints.mail_sources import _sanitize_for_log + + assert "\n" not in _sanitize_for_log("hello\nworld") + + def test_strips_carriage_return(self): + from app.api.api_v1.endpoints.mail_sources import _sanitize_for_log + + assert "\r" not in _sanitize_for_log("foo\rbar") + + def test_integer_is_safe(self): + from app.api.api_v1.endpoints.mail_sources import _sanitize_for_log + + assert _sanitize_for_log(42) == "42" + + def test_normal_string_unchanged(self): + from app.api.api_v1.endpoints.mail_sources import _sanitize_for_log + + assert _sanitize_for_log("example.com") == "example.com" + + +# --------------------------------------------------------------------------- +# Source-to-response helper (password masking) +# --------------------------------------------------------------------------- + + +class TestSourceToResponse: + """Tests for the _source_to_response password-masking helper.""" + + def test_password_is_redacted_when_set(self, db_session: Session): + from app.api.api_v1.endpoints.mail_sources import _source_to_response + + source = MailSource(name="Redact Test", method="IMAP", password="plaintext") + db_session.add(source) + db_session.commit() + db_session.refresh(source) + + response = _source_to_response(source) + assert response.password == "**redacted**" + + def test_password_is_none_when_not_set(self, db_session: Session): + from app.api.api_v1.endpoints.mail_sources import _source_to_response + + source = MailSource(name="No Password", method="IMAP") + db_session.add(source) + db_session.commit() + db_session.refresh(source) + + response = _source_to_response(source) + assert response.password is None + + +# --------------------------------------------------------------------------- +# Pytest marker to avoid warnings for test methods without assertions +# --------------------------------------------------------------------------- +pytestmark = pytest.mark.usefixtures("_reset_report_store")