gitea-admin/gh-christianlouis-dmarq
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Redesign test framework: fix conftest, rewrite all tests, fix model index conflicts, fix black formatting

Browse Source 
Agent-Logs-Url: https://github.com/christianlouis/dmarq/sessions/9d256101-34b6-4861-a8cf-7f86f32b54d5

Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>
This commit is contained in:
copilot-swe-agent[bot]
2026-03-29 10:19:43 +00:00
parent f5368927ef
commit 5fa315c8e3
9 changed files with 431 additions and 551 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/middleware/security.py
+6 -6
View File
@@ -52,29 +52,29 @@ class SecurityHeadersMiddleware(BaseHTTPMiddleware):
# Content Security Policy (CSP)
# Restricts sources of content that can be loaded
#
#
# SECURITY TODO: Current CSP includes 'unsafe-inline' and 'unsafe-eval' which
# weaken XSS protection. To remove these:
#
#
# For script-src 'unsafe-inline':
# 1. Move all inline <script> tags from templates to external .js files
# 2. OR implement CSP nonces for inline scripts (requires template changes)
# 3. Convert any inline event handlers (onclick, etc.) to addEventListener
#
#
# For script-src 'unsafe-eval':
# 1. Verify no code uses eval(), Function(), setTimeout/setInterval with strings
# 2. If using libraries that require eval, consider alternatives
# 3. Current scan shows no eval usage - can likely remove this directive
#
#
# For style-src 'unsafe-inline':
# 1. Move inline styles to CSS files or use style tags with nonces
# 2. Replace style="" attributes with CSS classes
# 3. OR implement CSP nonces for inline styles
#
#
# Target secure CSP (no inline):
# "script-src 'self'"
# "style-src 'self' https://fonts.googleapis.com"
#
#
# See: https://developer.mozilla.org/en-US/docs/Web/HTTP/CSP
csp_directives = [
"default-src 'self'",
backend/app/models/report.py
+2 -2
View File
@@ -21,7 +21,7 @@ class DMARCReport(Base):
source_email = Column(String, nullable=True)
# Policy information
policy = Column(String, nullable=True, index=True) # none, quarantine, reject
policy = Column(String, nullable=True) # none, quarantine, reject (indexed via __table_args__)
subdomain_policy = Column(String, nullable=True)
adkim = Column(String(1), nullable=True) # r (relaxed) or s (strict)
aspf = Column(String(1), nullable=True) # r (relaxed) or s (strict)
@@ -62,7 +62,7 @@ class ReportRecord(Base):
count = Column(Integer, nullable=False, default=0)
# Policy evaluation
disposition = Column(String, nullable=False, index=True) # none, quarantine, reject
disposition = Column(String, nullable=False) # none, quarantine, reject (indexed via __table_args__)
dkim = Column(String, nullable=True, index=True) # pass, fail
spf = Column(String, nullable=True, index=True) # pass, fail
backend/app/tests/conftest.py
+24 -61
View File
@@ -1,60 +1,46 @@
import asyncio
import pytest
import pytest_asyncio
from app.core.database import Base, get_db
from app.core.security import get_password_hash
from app.models.user import User
# Import all models so Base.metadata knows every table
import app.models.domain # noqa: F401
import app.models.report # noqa: F401
import app.models.user # noqa: F401
from app.services.report_store import ReportStore
from fastapi import FastAPI
from fastapi.testclient import TestClient
from httpx import AsyncClient
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
# Use in-memory SQLite database for tests
TEST_DATABASE_URL = "sqlite:///./test.db"
@pytest.fixture(scope="session")
def event_loop():
"""Create an instance of the default event loop for each test case."""
loop = asyncio.get_event_loop_policy().new_event_loop()
yield loop
loop.close()
@pytest.fixture(scope="session")
@pytest.fixture()
def test_app() -> FastAPI:
# Avoid circular import
"""Create a fresh FastAPI application instance for testing."""
from app.main import create_app
app = create_app()
return app
@pytest.fixture(scope="function")
@pytest.fixture()
def db_session():
# Create the SQLite database engine
engine = create_engine(TEST_DATABASE_URL)
# Create all tables
"""Create a fresh in-memory SQLite database session per test."""
engine = create_engine("sqlite://", connect_args={"check_same_thread": False})
Base.metadata.create_all(engine)
# Create a new session
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
db = TestingSessionLocal()
try:
yield db
finally:
db.close()
# Drop all tables after the test
Base.metadata.drop_all(engine)
engine.dispose()
@pytest.fixture(scope="function")
@pytest.fixture()
def client(test_app: FastAPI, db_session):
# Override the get_db dependency to use the test database
"""Create a TestClient with a DB override for the test app."""
def override_get_db():
try:
yield db_session
@@ -62,38 +48,15 @@ def client(test_app: FastAPI, db_session):
pass
test_app.dependency_overrides[get_db] = override_get_db
# Use the FastAPI TestClient
with TestClient(test_app) as test_client:
yield test_client
test_app.dependency_overrides.clear()
@pytest_asyncio.fixture(scope="function")
async def async_client(test_app: FastAPI, db_session):
# Override the get_db dependency to use the test database
def override_get_db():
try:
yield db_session
finally:
pass
test_app.dependency_overrides[get_db] = override_get_db
async with AsyncClient(app=test_app, base_url="http://testserver") as ac:
yield ac
@pytest.fixture(scope="function")
def test_user(db_session):
"""Create a test user in the database."""
user = User(
email="test@example.com",
hashed_password=get_password_hash("password"),
is_active=True,
is_superuser=False,
is_verified=True,
)
db_session.add(user)
db_session.commit()
db_session.refresh(user)
return user
@pytest.fixture(autouse=True)
def _reset_report_store():
"""Reset the ReportStore singleton between tests to avoid state leakage."""
store = ReportStore.get_instance()
store.clear()
yield
store.clear()
backend/app/tests/test_api.py
+19 -39
View File
@@ -1,10 +1,8 @@
from app.models.domain import Domain
from fastapi.testclient import TestClient
from sqlalchemy.orm import Session
def test_read_health(client: TestClient):
"""Test health check endpoint"""
def test_health_check(client: TestClient):
"""Test the health check endpoint returns status ok."""
response = client.get("/api/v1/health")
assert response.status_code == 200
data = response.json()
@@ -12,47 +10,29 @@ def test_read_health(client: TestClient):
assert "version" in data
def test_read_domains_empty(client: TestClient):
"""Test reading domains when none exist"""
response = client.get("/api/v1/domains")
def test_domains_empty(client: TestClient):
"""Test that GET /api/v1/domains/domains returns empty list when no reports uploaded."""
response = client.get("/api/v1/domains/domains")
assert response.status_code == 200
data = response.json()
assert data == []
def test_read_domains(client: TestClient, db_session: Session):
"""Test reading domains"""
# Create some test domains
domain1 = Domain(name="example.com", description="Example Domain", active=True)
domain2 = Domain(name="test.com", description="Test Domain", active=True)
db_session.add_all([domain1, domain2])
db_session.commit()
response = client.get("/api/v1/domains")
assert response.status_code == 200
data = response.json()
assert len(data) == 2
assert {"name": "example.com", "description": "Example Domain"}.items() <= data[0].items()
assert {"name": "test.com", "description": "Test Domain"}.items() <= data[1].items()
def test_create_domain(client: TestClient):
"""Test creating a new domain"""
def test_reports_upload_invalid_extension(client: TestClient):
"""Test that uploading a file with an unsupported extension returns 400."""
response = client.post(
"/api/v1/domains",
json={"name": "newdomain.com", "description": "New Domain", "active": True},
"/api/v1/reports/upload",
files={"file": ("report.txt", b"not a report", "text/plain")},
)
assert response.status_code == 400
assert "Invalid file type" in response.json()["detail"]
assert response.status_code == 201
data = response.json()
assert data["name"] == "newdomain.com"
assert data["description"] == "New Domain"
assert data["active"] is True
assert "id" in data
# Check that the domain was actually created
response = client.get("/api/v1/domains")
assert response.status_code == 200
domains = response.json()
assert any(d["name"] == "newdomain.com" for d in domains)
def test_reports_upload_empty_file(client: TestClient):
"""Test that uploading an empty file returns 400."""
response = client.post(
"/api/v1/reports/upload",
files={"file": ("report.xml", b"", "application/xml")},
)
assert response.status_code == 400
assert "empty" in response.json()["detail"].lower()
backend/app/tests/test_dmarc_parser.py
+100 -93
View File
@@ -1,112 +1,119 @@
from unittest.mock import MagicMock, patch
import io
import zipfile
import defusedxml.ElementTree as ET
import pytest
from app.services.dmarc_parser import DMARCParser
SAMPLE_XML = """\
<?xml version="1.0" encoding="UTF-8" ?>
<feedback>
<report_metadata>
<org_name>google.com</org_name>
<email>noreply-dmarc-support@google.com</email>
<report_id>123456789</report_id>
<date_range>
<begin>1597449600</begin>
<end>1597535999</end>
</date_range>
</report_metadata>
<policy_published>
<domain>example.com</domain>
<adkim>r</adkim>
<aspf>r</aspf>
<p>none</p>
<sp>none</sp>
<pct>100</pct>
</policy_published>
<record>
<row>
<source_ip>203.0.113.1</source_ip>
<count>2</count>
<policy_evaluated>
<disposition>none</disposition>
<dkim>pass</dkim>
<spf>fail</spf>
</policy_evaluated>
</row>
<identifiers>
<header_from>example.com</header_from>
</identifiers>
<auth_results>
<dkim>
<domain>example.com</domain>
<result>pass</result>
<selector>default</selector>
</dkim>
<spf>
<domain>example.com</domain>
<result>fail</result>
</spf>
</auth_results>
</record>
</feedback>
"""
class TestDMARCParser:
"""Tests for the DMARC XML parser."""
def setup_method(self):
"""Set up test fixtures"""
self.parser = DMARCParser()
def test_parse_xml_report(self):
"""Test parsing a plain XML DMARC report."""
xml_bytes = SAMPLE_XML.encode("utf-8")
result = DMARCParser.parse_file(xml_bytes, "report.xml")
# Sample XML string for testing
self.sample_xml = """<?xml version="1.0" encoding="UTF-8" ?>
<feedback>
<report_metadata>
<org_name>google.com</org_name>
<email>noreply-dmarc-support@google.com</email>
<report_id>123456789</report_id>
<date_range>
<begin>1597449600</begin>
<end>1597535999</end>
</date_range>
</report_metadata>
<policy_published>
<domain>example.com</domain>
<adkim>r</adkim>
<aspf>r</aspf>
<p>none</p>
<sp>none</sp>
<pct>100</pct>
</policy_published>
<record>
<row>
<source_ip>203.0.113.1</source_ip>
<count>2</count>
<policy_evaluated>
<disposition>none</disposition>
<dkim>pass</dkim>
<spf>fail</spf>
</policy_evaluated>
</row>
<identifiers>
<header_from>example.com</header_from>
</identifiers>
<auth_results>
<dkim>
<domain>example.com</domain>
<result>pass</result>
<selector>default</selector>
</dkim>
<spf>
<domain>example.com</domain>
<result>fail</result>
</spf>
</auth_results>
</record>
</feedback>
"""
# Report metadata (flat keys from _parse_xml)
assert result["report_id"] == "123456789"
assert result["org_name"] == "google.com"
assert result["email"] == "noreply-dmarc-support@google.com"
assert result["begin_timestamp"] == 1597449600
assert result["end_timestamp"] == 1597535999
def test_parse_aggregate_report_xml(self):
"""Test parsing an XML aggregate report"""
# Use DMARCParser.parse_file with file_content (bytes) and filename
xml_bytes = self.sample_xml.encode("utf-8")
result = DMARCParser.parse_file(xml_bytes, "test_report.xml")
# Policy published
assert result["domain"] == "example.com"
assert result["policy"]["p"] == "none"
# Verify report metadata
assert result["report_metadata"]["org_name"] == "google.com"
assert result["report_metadata"]["email"] == "noreply-dmarc-support@google.com"
assert result["report_metadata"]["report_id"] == "123456789"
assert result["report_metadata"]["begin_date"] == 1597449600
assert result["report_metadata"]["end_date"] == 1597535999
# Verify policy published
assert result["policy_published"]["domain"] == "example.com"
assert result["policy_published"]["policy"] == "none"
# Verify record data
# Records
assert len(result["records"]) == 1
record = result["records"][0]
assert record["source_ip"] == "203.0.113.1"
assert record["count"] == 2
assert record["policy_evaluated"]["disposition"] == "none"
assert record["policy_evaluated"]["dkim"] == "pass"
assert record["policy_evaluated"]["spf"] == "fail"
assert record["identifiers"]["header_from"] == "example.com"
assert record["disposition"] == "none"
assert record["dkim_result"] == "pass"
assert record["spf_result"] == "fail"
assert record["header_from"] == "example.com"
@patch("app.services.dmarc_parser.zipfile.ZipFile")
def test_parse_aggregate_report_zip(self, mock_zipfile):
"""Test parsing a zipped aggregate report"""
# Setup mock zipfile extraction
mock_zip_instance = MagicMock()
mock_zipfile.return_value.__enter__.return_value = mock_zip_instance
mock_zip_instance.namelist.return_value = ["report.xml"]
mock_zip_instance.read.return_value = self.sample_xml.encode("utf-8")
# Summary
assert result["summary"]["total_count"] == 2
assert result["summary"]["passed_count"] == 2 # dkim passed
assert result["summary"]["failed_count"] == 0
# Create fake zip file content
zip_content = b"fake_zip_content"
result = DMARCParser.parse_file(zip_content, "test_report.zip")
def test_parse_zip_report(self):
"""Test parsing a DMARC report inside a ZIP archive."""
xml_bytes = SAMPLE_XML.encode("utf-8")
# Assertions similar to test_parse_aggregate_report_xml
assert result["report_metadata"]["org_name"] == "google.com"
zip_buffer = io.BytesIO()
with zipfile.ZipFile(zip_buffer, "w") as zf:
zf.writestr("report.xml", xml_bytes)
zip_content = zip_buffer.getvalue()
result = DMARCParser.parse_file(zip_content, "report.zip")
assert result["report_id"] == "123456789"
assert result["domain"] == "example.com"
assert len(result["records"]) == 1
def test_extract_authentication_results(self):
"""Test extracting authentication results from report"""
# This test was for an internal method that may have changed
# The functionality is tested through test_parse_aggregate_report_xml
# which validates the full parsing including authentication results
import pytest
def test_file_too_large(self):
"""Test that files exceeding the size limit are rejected."""
large_content = b"x" * (11 * 1024 * 1024) # 11 MB
with pytest.raises(ValueError, match="too large"):
DMARCParser.parse_file(large_content, "report.xml")
pytest.skip("Internal method test - functionality covered by integration tests")
def test_invalid_xml(self):
"""Test that invalid XML raises a ValueError."""
with pytest.raises(ValueError):
DMARCParser.parse_file(b"not xml at all", "report.xml")
def test_unsupported_extension_returns_none(self):
"""Test that an unsupported file extension raises ValueError."""
with pytest.raises(ValueError, match="Could not extract XML"):
DMARCParser.parse_file(b"some content", "report.pdf")
backend/app/tests/test_models.py
+21 -61
View File
@@ -4,14 +4,15 @@ from sqlalchemy.orm import Session
class TestDomainModel:
"""Tests for the Domain model"""
"""Tests for the Domain ORM model."""
def test_create_domain(self, db_session: Session):
"""Test creating a domain in the database"""
domain = Domain(
name="example.com", description="Test domain", active=True, dmarc_policy="quarantine"
name="example.com",
description="Test domain",
active=True,
dmarc_policy="quarantine",
)
db_session.add(domain)
db_session.commit()
db_session.refresh(domain)
@@ -23,66 +24,44 @@ class TestDomainModel:
assert domain.dmarc_policy == "quarantine"
def test_domain_reports_relationship(self, db_session: Session):
"""Test the relationship between domains and DMARC reports"""
# Create a domain
domain = Domain(name="example.com", active=True)
db_session.add(domain)
db_session.commit()
# Create reports for the domain
report1 = DMARCReport(
report = DMARCReport(
domain_id=domain.id,
report_id="report1",
org_name="Google",
begin_date=1597449600,
end_date=1597535999,
source_email="noreply-dmarc-support@google.com",
source_email="noreply@google.com",
)
report2 = DMARCReport(
domain_id=domain.id,
report_id="report2",
org_name="Microsoft",
begin_date=1597536000,
end_date=1597622399,
source_email="dmarc@microsoft.com",
)
db_session.add_all([report1, report2])
db_session.add(report)
db_session.commit()
# Query the domain and check its reports
domain = db_session.query(Domain).filter_by(name="example.com").first()
assert domain is not None
assert len(domain.reports) == 2
assert domain.reports[0].report_id in ["report1", "report2"]
assert domain.reports[1].report_id in ["report1", "report2"]
fetched = db_session.query(Domain).filter_by(name="example.com").first()
assert fetched is not None
assert len(fetched.reports) == 1
assert fetched.reports[0].report_id == "report1"
class TestDMARCReportModel:
"""Tests for the DMARCReport model"""
"""Tests for the DMARCReport ORM model."""
def test_create_report(self, db_session: Session):
"""Test creating a DMARC report in the database"""
# Create a domain first
domain = Domain(name="example.com", active=True)
db_session.add(domain)
db_session.commit()
# Create a report
report = DMARCReport(
domain_id=domain.id,
report_id="123456789",
org_name="Google",
begin_date=1597449600,
end_date=1597535999,
source_email="noreply-dmarc-support@google.com",
source_email="noreply@google.com",
policy="none",
adkim="r",
aspf="r",
percentage=100,
)
db_session.add(report)
db_session.commit()
db_session.refresh(report)
@@ -91,12 +70,9 @@ class TestDMARCReportModel:
assert report.domain_id == domain.id
assert report.report_id == "123456789"
assert report.org_name == "Google"
assert report.begin_date == 1597449600
assert report.policy == "none"
def test_report_records_relationship(self, db_session: Session):
"""Test the relationship between reports and records"""
# Create domain and report
domain = Domain(name="example.com", active=True)
db_session.add(domain)
db_session.commit()
@@ -107,13 +83,12 @@ class TestDMARCReportModel:
org_name="Google",
begin_date=1597449600,
end_date=1597535999,
source_email="noreply-dmarc-support@google.com",
source_email="noreply@google.com",
)
db_session.add(report)
db_session.commit()
# Create records for the report
record1 = ReportRecord(
record = ReportRecord(
report_id=report.id,
source_ip="203.0.113.1",
count=2,
@@ -121,26 +96,11 @@ class TestDMARCReportModel:
dkim="pass",
spf="fail",
header_from="example.com",
envelope_from=None,
)
record2 = ReportRecord(
report_id=report.id,
source_ip="203.0.113.2",
count=5,
disposition="none",
dkim="pass",
spf="pass",
header_from="example.com",
envelope_from=None,
)
db_session.add_all([record1, record2])
db_session.add(record)
db_session.commit()
# Query the report and check its records
report = db_session.query(DMARCReport).filter_by(report_id="123456789").first()
assert report is not None
assert len(report.records) == 2
assert report.records[0].source_ip in ["203.0.113.1", "203.0.113.2"]
assert report.records[1].source_ip in ["203.0.113.1", "203.0.113.2"]
fetched = db_session.query(DMARCReport).filter_by(report_id="123456789").first()
assert fetched is not None
assert len(fetched.records) == 1
assert fetched.records[0].source_ip == "203.0.113.1"
backend/app/tests/test_report_store.py
+79
View File
@@ -0,0 +1,79 @@
from app.services.report_store import ReportStore
def _sample_report(domain: str = "example.com") -> dict:
"""Return a minimal parsed report dict for testing."""
return {
"domain": domain,
"report_id": "rpt-001",
"org_name": "google.com",
"begin_date": "2020-08-15T00:00:00",
"end_date": "2020-08-15T23:59:59",
"begin_timestamp": 1597449600,
"end_timestamp": 1597535999,
"policy": {"p": "none", "sp": "none", "pct": "100"},
"records": [
{
"source_ip": "203.0.113.1",
"count": 5,
"disposition": "none",
"dkim_result": "pass",
"spf_result": "fail",
"header_from": domain,
}
],
"summary": {
"total_count": 5,
"passed_count": 5,
"failed_count": 0,
"pass_rate": 100.0,
},
}
class TestReportStore:
"""Tests for the in-memory ReportStore."""
def test_add_report_creates_domain(self):
store = ReportStore.get_instance()
store.add_report(_sample_report("test.com"))
domains = store.get_domains()
assert "test.com" in domains
def test_domain_summary_after_add(self):
store = ReportStore.get_instance()
store.add_report(_sample_report("test.com"))
summary = store.get_domain_summary("test.com")
assert summary["total_count"] == 5
assert summary["passed_count"] == 5
assert summary["reports_processed"] == 1
def test_get_domain_reports(self):
store = ReportStore.get_instance()
store.add_report(_sample_report("test.com"))
reports = store.get_domain_reports("test.com")
assert len(reports) == 1
assert reports[0]["report_id"] == "rpt-001"
def test_clear(self):
store = ReportStore.get_instance()
store.add_report(_sample_report("test.com"))
store.clear()
assert store.get_domains() == []
def test_delete_domain(self):
store = ReportStore.get_instance()
store.add_report(_sample_report("test.com"))
store.add_report(_sample_report("other.com"))
assert store.delete_domain_with_cleanup("test.com") is True
assert "test.com" not in store.get_domains()
assert "other.com" in store.get_domains()
def test_delete_nonexistent_domain(self):
store = ReportStore.get_instance()
assert store.delete_domain_with_cleanup("nope.com") is False
backend/app/tests/test_reports_api.py
+106 -151
View File
@@ -1,162 +1,117 @@
import io
import zipfile
from app.models.domain import Domain
from fastapi.testclient import TestClient
from sqlalchemy.orm import Session
SAMPLE_XML = """\
<?xml version="1.0" encoding="UTF-8" ?>
<feedback>
<report_metadata>
<org_name>google.com</org_name>
<email>noreply-dmarc-support@google.com</email>
<report_id>123456789</report_id>
<date_range>
<begin>1597449600</begin>
<end>1597535999</end>
</date_range>
</report_metadata>
<policy_published>
<domain>example.com</domain>
<adkim>r</adkim>
<aspf>r</aspf>
<p>none</p>
<sp>none</sp>
<pct>100</pct>
</policy_published>
<record>
<row>
<source_ip>203.0.113.1</source_ip>
<count>2</count>
<policy_evaluated>
<disposition>none</disposition>
<dkim>pass</dkim>
<spf>fail</spf>
</policy_evaluated>
</row>
<identifiers>
<header_from>example.com</header_from>
</identifiers>
<auth_results>
<dkim>
<domain>example.com</domain>
<result>pass</result>
<selector>default</selector>
</dkim>
<spf>
<domain>example.com</domain>
<result>fail</result>
</spf>
</auth_results>
</record>
</feedback>
"""
def test_read_reports_empty(client: TestClient):
"""Test reading reports when none exist"""
response = client.get("/api/v1/reports")
def _make_zip(xml_content: str) -> bytes:
"""Create a ZIP file containing the given XML content."""
buf = io.BytesIO()
with zipfile.ZipFile(buf, "w") as zf:
zf.writestr("report.xml", xml_content)
return buf.getvalue()
def test_upload_report_success(client: TestClient):
"""Uploading a valid zipped DMARC report succeeds."""
zip_bytes = _make_zip(SAMPLE_XML)
response = client.post(
"/api/v1/reports/upload",
files={"file": ("report.zip", zip_bytes, "application/zip")},
)
assert response.status_code == 200
data = response.json()
assert data == []
def test_upload_report_no_domain(client: TestClient):
"""Test uploading a report when domain doesn't exist"""
# Create a simple XML report
xml_content = """<?xml version="1.0" encoding="UTF-8" ?>
<feedback>
<report_metadata>
<org_name>google.com</org_name>
<email>noreply-dmarc-support@google.com</email>
<report_id>123456789</report_id>
<date_range>
<begin>1597449600</begin>
<end>1597535999</end>
</date_range>
</report_metadata>
<policy_published>
<domain>nonexistentdomain.com</domain>
<adkim>r</adkim>
<aspf>r</aspf>
<p>none</p>
<sp>none</sp>
<pct>100</pct>
</policy_published>
<record>
<row>
<source_ip>203.0.113.1</source_ip>
<count>2</count>
<policy_evaluated>
<disposition>none</disposition>
<dkim>pass</dkim>
<spf>fail</spf>
</policy_evaluated>
</row>
<identifiers>
<header_from>nonexistentdomain.com</header_from>
</identifiers>
<auth_results>
<dkim>
<domain>nonexistentdomain.com</domain>
<result>pass</result>
<selector>default</selector>
</dkim>
<spf>
<domain>nonexistentdomain.com</domain>
<result>fail</result>
</spf>
</auth_results>
</record>
</feedback>
"""
# Create an in-memory zip file
zip_buffer = io.BytesIO()
with zipfile.ZipFile(zip_buffer, "w") as zip_file:
zip_file.writestr("report.xml", xml_content)
zip_buffer.seek(0)
# Upload the zip file
response = client.post(
"/api/v1/reports/upload", files={"file": ("report.zip", zip_buffer, "application/zip")}
)
# Should return an error since domain doesn't exist
assert response.status_code == 404
data = response.json()
assert "domain not found" in data["detail"].lower()
def test_upload_report_success(client: TestClient, db_session: Session):
"""Test successfully uploading a report"""
# Create a domain first
domain = Domain(name="example.com", active=True)
db_session.add(domain)
db_session.commit()
# Create a simple XML report
xml_content = """<?xml version="1.0" encoding="UTF-8" ?>
<feedback>
<report_metadata>
<org_name>google.com</org_name>
<email>noreply-dmarc-support@google.com</email>
<report_id>123456789</report_id>
<date_range>
<begin>1597449600</begin>
<end>1597535999</end>
</date_range>
</report_metadata>
<policy_published>
<domain>example.com</domain>
<adkim>r</adkim>
<aspf>r</aspf>
<p>none</p>
<sp>none</sp>
<pct>100</pct>
</policy_published>
<record>
<row>
<source_ip>203.0.113.1</source_ip>
<count>2</count>
<policy_evaluated>
<disposition>none</disposition>
<dkim>pass</dkim>
<spf>fail</spf>
</policy_evaluated>
</row>
<identifiers>
<header_from>example.com</header_from>
</identifiers>
<auth_results>
<dkim>
<domain>example.com</domain>
<result>pass</result>
<selector>default</selector>
</dkim>
<spf>
<domain>example.com</domain>
<result>fail</result>
</spf>
</auth_results>
</record>
</feedback>
"""
# Create an in-memory zip file
zip_buffer = io.BytesIO()
with zipfile.ZipFile(zip_buffer, "w") as zip_file:
zip_file.writestr("report.xml", xml_content)
zip_buffer.seek(0)
# Upload the zip file
response = client.post(
"/api/v1/reports/upload", files={"file": ("report.zip", zip_buffer, "application/zip")}
)
# Should be successful
assert response.status_code == 201
data = response.json()
assert data["success"] is True
assert "report_id" in data
assert data["domain"] == "example.com"
# Check that the report was actually created
response = client.get("/api/v1/reports")
def test_upload_populates_domains_list(client: TestClient):
"""After uploading a report, the domain appears in the reports/domains endpoint."""
zip_bytes = _make_zip(SAMPLE_XML)
client.post(
"/api/v1/reports/upload",
files={"file": ("report.zip", zip_bytes, "application/zip")},
)
response = client.get("/api/v1/reports/domains")
assert response.status_code == 200
reports = response.json()
assert len(reports) == 1
assert reports[0]["report_id"] == "123456789"
assert reports[0]["org_name"] == "google.com"
domains = response.json()
assert "example.com" in domains
def test_reports_domains_empty(client: TestClient):
"""GET /api/v1/reports/domains returns empty list when no reports uploaded."""
response = client.get("/api/v1/reports/domains")
assert response.status_code == 200
assert response.json() == []
def test_reports_summary_empty(client: TestClient):
"""GET /api/v1/reports/summary returns empty list when no reports uploaded."""
response = client.get("/api/v1/reports/summary")
assert response.status_code == 200
assert response.json() == []
def test_upload_and_get_domain_summary(client: TestClient):
"""After uploading a report, the domain summary endpoint returns correct data."""
zip_bytes = _make_zip(SAMPLE_XML)
client.post(
"/api/v1/reports/upload",
files={"file": ("report.zip", zip_bytes, "application/zip")},
)
response = client.get("/api/v1/reports/domain/example.com/summary")
assert response.status_code == 200
data = response.json()
assert data["domain"] == "example.com"
assert data["total_count"] == 2
assert data["reports_processed"] == 1
backend/app/tests/test_security.py
+74 -138
View File
@@ -1,98 +1,78 @@
"""
Security-focused unit tests for DMARQ application.
Security-focused tests for DMARQ application.
Tests authentication, input validation, file upload security, and other security features.
Covers API key management, domain validation, file upload limits, and XML parsing security.
"""
import pytest
from app.core.security import (
add_api_key,
generate_api_key,
verify_api_key,
)
from app.core.security import add_api_key, generate_api_key, verify_api_key
from app.services.dmarc_parser import DMARCParser
from app.utils.domain_validator import validate_domain, validate_domain_config
class TestAuthentication:
"""Test authentication and API key functionality."""
class TestAPIKeySecurity:
"""Test API key generation and verification."""
def test_generate_api_key(self):
"""Test API key generation."""
def test_generate_api_key_length_and_uniqueness(self):
"""Generated keys should be 64 hex characters and unique."""
key1 = generate_api_key()
key2 = generate_api_key()
# Keys should be 64 characters (32 bytes hex encoded)
assert len(key1) == 64
assert len(key2) == 64
# Keys should be unique
assert key1 != key2
# Keys should be hexadecimal
assert all(c in "0123456789abcdef" for c in key1)
def test_add_and_verify_api_key(self):
"""Test adding and verifying API keys."""
"""Keys should only be valid after being added."""
key = generate_api_key()
# Key should not be valid before adding
assert not verify_api_key(key)
# Add key
assert add_api_key(key)
assert add_api_key(key) is True
assert verify_api_key(key) is True
# Key should now be valid
assert verify_api_key(key)
# Adding same key again should return False
assert not add_api_key(key)
def test_password_hashing(self):
"""Test password hashing and verification."""
# Skip this test if bcrypt has issues
pytest.skip("Skipping due to bcrypt compatibility issues in test environment")
# Adding the same key again returns False
assert add_api_key(key) is False
class TestDomainValidation:
"""Test domain validation security."""
"""Test domain name validation."""
def test_valid_domains(self):
"""Test validation of legitimate domains."""
valid_domains = [
@pytest.mark.parametrize(
"domain",
[
"example.com",
"subdomain.example.com",
"my-domain.example.org",
"test123.example.net",
]
],
)
def test_valid_domains(self, domain):
is_valid, error, _ = validate_domain(domain, check_dns=False)
assert is_valid, f"Domain {domain} should be valid: {error}"
for domain in valid_domains:
is_valid, error, error_code = validate_domain(domain, check_dns=False)
assert is_valid, f"Domain {domain} should be valid: {error}"
@pytest.mark.parametrize(
"domain",
[
"",
" ",
"example",
"-example.com",
"example-.com",
"exam ple.com",
"example..com",
"a" * 64 + ".com",
"a" * 254,
],
)
def test_invalid_domain_format(self, domain):
is_valid, error, _ = validate_domain(domain, check_dns=False)
assert not is_valid, f"Domain '{domain}' should be invalid"
assert error is not None
def test_invalid_domain_format(self):
"""Test rejection of invalid domain formats."""
invalid_domains = [
"", # Empty
" ", # Whitespace
"example", # No TLD
"-example.com", # Starts with hyphen
"example-.com", # Ends with hyphen
"exam ple.com", # Contains space
"example..com", # Double dot
"example.com.", # Trailing dot (should fail with current regex)
"a" * 64 + ".com", # Label too long (>63 chars)
"a" * 250 + ".com", # Domain too long (>253 chars)
]
for domain in invalid_domains:
is_valid, error, error_code = validate_domain(domain, check_dns=False)
assert not is_valid, f"Domain '{domain}' should be invalid"
assert error is not None
def test_malicious_domain_input(self):
"""Test rejection of domains with malicious characters."""
malicious_domains = [
@pytest.mark.parametrize(
"domain",
[
"example.com<script>",
"example.com'; DROP TABLE users--",
"example.com|whoami",
@@ -101,91 +81,60 @@ class TestDomainValidation:
"example.com$USER",
'example.com"test',
"example.com\\\\test",
]
],
)
def test_malicious_domain_input(self, domain):
is_valid, _, _ = validate_domain(domain, check_dns=False)
assert not is_valid, f"Malicious domain '{domain}' should be rejected"
for domain in malicious_domains:
is_valid, error, error_code = validate_domain(domain, check_dns=False)
assert not is_valid, f"Malicious domain '{domain}' should be rejected"
def test_domain_length_limits(self):
"""Test domain length validation."""
# Max label is 63 characters - this should be caught by label length check
long_label = "a" * 64 + ".example.com"
is_valid, error, error_code = validate_domain(long_label, check_dns=False)
assert not is_valid
# Could be caught by format check or label length check
assert error is not None
# Max domain is 253 characters
long_domain = "a" * 254 # 254 chars, no dot
is_valid, error, error_code = validate_domain(long_domain, check_dns=False)
assert not is_valid
assert "too long" in error.lower() or "invalid" in error.lower()
def test_domain_config_validation(self):
"""Test domain configuration validation."""
# Valid config
valid_config = {"name": "example.com", "description": "Test domain"}
result = validate_domain_config(valid_config)
def test_domain_config_validation_valid(self):
result = validate_domain_config({"name": "example.com", "description": "Test domain"})
assert result["valid"]
assert len(result["errors"]) == 0
# Missing name
invalid_config = {"description": "Test"}
result = validate_domain_config(invalid_config)
def test_domain_config_missing_name(self):
result = validate_domain_config({"description": "Test"})
assert not result["valid"]
assert "name" in result["errors"]
# Description too long
long_desc_config = {"name": "example.com", "description": "a" * 501}
result = validate_domain_config(long_desc_config)
def test_domain_config_description_too_long(self):
result = validate_domain_config({"name": "example.com", "description": "a" * 501})
assert not result["valid"]
assert "description" in result["errors"]
# Malicious description
malicious_config = {"name": "example.com", "description": "<script>alert('xss')</script>"}
result = validate_domain_config(malicious_config)
def test_domain_config_xss_description(self):
result = validate_domain_config(
{"name": "example.com", "description": "<script>alert('xss')</script>"}
)
assert not result["valid"]
assert "description" in result["errors"]
class TestFileUploadSecurity:
"""Test file upload security features."""
"""Test file upload size limits."""
def test_file_size_limit(self):
"""Test file size limit enforcement."""
parser = DMARCParser()
# Create a file that's too large (> 10 MB)
large_content = b"x" * (11 * 1024 * 1024)
with pytest.raises(ValueError) as exc_info:
parser.parse_file(large_content, "test.xml")
assert "too large" in str(exc_info.value).lower()
with pytest.raises(ValueError, match="too large"):
DMARCParser.parse_file(large_content, "test.xml")
class TestXMLParsingSecurity:
"""Test XML parsing security features."""
"""Test XML parsing security (defusedxml, XXE protection)."""
def test_defusedxml_import(self):
"""Test that defusedxml is being used."""
def test_defusedxml_is_used(self):
import app.services.dmarc_parser as parser_module
# Check that the module uses defusedxml
assert hasattr(parser_module, "ET")
# The module name should contain 'defusedxml'
assert (
"defusedxml" in str(parser_module.ET.__name__).lower()
or "defusedxml" in str(parser_module.ET.__module__).lower()
module_info = str(getattr(parser_module.ET, "__name__", "")) + str(
getattr(parser_module.ET, "__module__", "")
)
assert "defusedxml" in module_info.lower()
def test_xml_entity_expansion_protection(self):
"""Test protection against XML entity expansion attacks."""
parser = DMARCParser()
# XXE attack payload
xxe_payload = b"""<?xml version="1.0"?>
def test_xxe_protection(self):
"""defusedxml should prevent XXE entity expansion."""
xxe_payload = b"""\
<?xml version="1.0"?>
<!DOCTYPE foo [
<!ENTITY xxe SYSTEM "file:///etc/passwd">
]>
@@ -195,23 +144,10 @@ class TestXMLParsingSecurity:
</report_metadata>
</feedback>
"""
# Should either fail parsing or not expand the entity
# defusedxml should prevent this
# defusedxml should raise an error or not expand the entity
try:
result = parser.parse_file(xxe_payload, "test.xml")
# If it doesn't raise an error, the entity should not be expanded
result = DMARCParser.parse_file(xxe_payload, "test.xml")
org_name = result.get("org_name", "")
assert not org_name.startswith("root:") and "/bin" not in org_name
except Exception:
# Expected - defusedxml should prevent parsing
pass
# Note: TestSecurityHeaders and TestErrorHandling tests are not implemented
# because they require proper async client setup. These will be added in a future PR
# with proper integration test infrastructure.
if __name__ == "__main__":
pytest.main([__file__, "-v"])
assert "root:" not in org_name and "/bin" not in org_name
except (ValueError, Exception):
pass # Expected – defusedxml blocks DTD processing