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

Fix pylint warnings: logging, globals, exceptions, imports, duplicates

Browse Source 
Agent-Logs-Url: https://github.com/christianlouis/dmarq/sessions/576427d4-4f6a-46d2-b75f-6862ecbcf526

Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>
This commit is contained in:
copilot-swe-agent[bot]
2026-03-29 11:21:35 +00:00
parent 6eb99da749
commit 50aa5bd5da
16 changed files with 182 additions and 213 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/api/api_v1/endpoints/domains.py
+6 -2
View File
@@ -248,9 +248,13 @@ async def get_domain_dns_records(domain_id: str = Path(..., title="The domain ID
# For Milestone 1, return mock DNS record data
# In a future milestone, this will be replaced with actual DNS lookups
mock_dmarc_record = (
"v=DMARC1; p=none; rua=mailto:dmarc@example.com;"
" ruf=mailto:forensic@example.com; pct=100"
)
return DNSRecordResponse(
dmarc=True,
dmarcRecord="v=DMARC1; p=none; rua=mailto:dmarc@example.com; ruf=mailto:forensic@example.com; pct=100",
dmarcRecord=mock_dmarc_record,
spf=True,
spfRecord="v=spf1 include:_spf.google.com include:spf.protection.outlook.com -all",
dkim=True,
@@ -299,7 +303,7 @@ async def get_domain_reports(
date = datetime.now() - timedelta(days=i)
date_str = date.strftime("%Y-%m-%d")
# TODO: Replace with actual historical data in future milestone
# TODO: Replace with actual historical data in future milestone # pylint: disable=fixme
# For now, generate mock data with variation for demonstration purposes
compliance_rate = random.uniform(80, 100) # nosec B311 - Mock data only
backend/app/api/api_v1/endpoints/reports.py
+12 -12
View File
@@ -47,16 +47,16 @@ def _validate_mime_type(file_content: bytes) -> None:
try:
mime_type = magic.from_buffer(file_content, mime=True)
if mime_type not in ALLOWED_MIME_TYPES:
logger.warning(f"Rejected file with MIME type: {mime_type}")
logger.warning("Rejected file with MIME type: %s", mime_type)
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Invalid file type. File must be XML, ZIP, or GZIP format.",
)
except HTTPException:
raise
except Exception as e:
except Exception as e: # pylint: disable=broad-exception-caught
# If magic fails, log but continue (fallback to extension check)
logger.warning(f"MIME type detection failed: {str(e)}")
logger.warning("MIME type detection failed: %s", str(e))
def _validate_upload_file(file: UploadFile, file_content: bytes) -> None:
@@ -89,15 +89,14 @@ def _handle_upload_value_error(filename: str, error_message: str) -> None:
Always raises — never returns.
"""
logger.error(f"ValueError processing report {filename}: {error_message}")
logger.error("ValueError processing report %s: %s", filename, error_message)
if "too large" in error_message.lower():
raise HTTPException(
status_code=status.HTTP_413_REQUEST_ENTITY_TOO_LARGE, detail="File too large"
)
elif "zip bomb" in error_message.lower():
if "zip bomb" in error_message.lower():
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Invalid archive file")
else:
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Invalid report format")
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Invalid report format")
class UploadResponse(BaseModel):
@@ -196,13 +195,13 @@ async def upload_report(file: UploadFile = File(...)):
except ValueError as e:
# Security: Sanitize error messages from parser
_handle_upload_value_error(file.filename, str(e))
except Exception as e:
except Exception as e: # pylint: disable=broad-exception-caught
# Security: Don't expose internal errors to client
logger.error(f"Unexpected error processing report {file.filename}: {str(e)}")
logger.error("Unexpected error processing report %s: %s", file.filename, str(e))
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="Error processing report. Please contact support if this persists.",
)
) from e
@router.get("/domains", response_model=List[str])
@@ -300,10 +299,11 @@ async def get_domain_reports_paginated(
if sort_field == "total_count":
all_reports.sort(
key=lambda r: r.get("summary", {}).get("total_count", 0), reverse=(sort_order == "desc")
key=lambda r: r.get("summary", {}).get("total_count", 0),
reverse=sort_order == "desc",
)
else:
all_reports.sort(key=lambda r: r.get(sort_field, ""), reverse=(sort_order == "desc"))
all_reports.sort(key=lambda r: r.get(sort_field, ""), reverse=sort_order == "desc")
# Apply pagination
total = len(all_reports)
backend/app/core/config.py
+7 -6
View File
@@ -5,7 +5,7 @@ from typing import List, Optional, Union
# Try to import from pydantic_settings first (newer versions)
try:
from pydantic import EmailStr, validator
from pydantic import EmailStr, validator # pylint: disable=ungrouped-imports
from pydantic_settings import BaseSettings
except ImportError:
# Fall back to older pydantic version
@@ -47,7 +47,7 @@ class Settings(BaseSettings):
CLOUDFLARE_ZONE_ID: Optional[str] = None
@validator("SECRET_KEY", pre=True, always=True)
def validate_secret_key(cls, v: Optional[str]) -> str:
def validate_secret_key(cls, v: Optional[str]) -> str: # pylint: disable=no-self-argument
"""Validate and generate SECRET_KEY if not provided."""
# Default insecure key that should never be used
DEFAULT_INSECURE_KEY = "CHANGE_THIS_TO_A_RANDOM_SECRET_IN_PRODUCTION"
@@ -66,17 +66,18 @@ class Settings(BaseSettings):
# Check if key is too short
if len(v) < 32:
logger.warning(
f"SECRET_KEY is too short ({len(v)} characters). "
"Recommended minimum is 32 characters for security."
"SECRET_KEY is too short (%s characters). "
"Recommended minimum is 32 characters for security.",
len(v),
)
return v
@validator("BACKEND_CORS_ORIGINS", pre=True)
def assemble_cors_origins(cls, v: Union[str, List[str]]) -> List[str]:
def assemble_cors_origins(cls, v: Union[str, List[str]]) -> List[str]: # pylint: disable=no-self-argument
if isinstance(v, str) and not v.startswith("["):
return [i.strip() for i in v.split(",")]
elif isinstance(v, (list, str)):
if isinstance(v, (list, str)):
return v
raise ValueError(v)
backend/app/core/security.py
+10 -11
View File
@@ -75,7 +75,7 @@ def add_api_key(api_key: str) -> bool:
if api_key in _api_keys:
return False
_api_keys.add(api_key)
logger.info(f"API key added (ends with: ...{api_key[-8:]})")
logger.info("API key added (ends with: ...%s)", api_key[-8:])
return True
@@ -92,7 +92,7 @@ def verify_api_key(api_key: str) -> bool:
return api_key in _api_keys
async def get_api_key(api_key_header: Optional[str] = Security(api_key_header)) -> str:
async def get_api_key(api_key_value: Optional[str] = Security(api_key_header)) -> str:
"""
Dependency to verify API key authentication.
@@ -105,24 +105,23 @@ async def get_api_key(api_key_header: Optional[str] = Security(api_key_header))
Raises:
HTTPException: If API key is missing or invalid
"""
if not api_key_header:
if not api_key_value:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Missing API key",
headers={"WWW-Authenticate": "ApiKey"},
)
if not verify_api_key(api_key_header):
logger.warning(
f"Invalid API key attempt: ...{api_key_header[-8:] if len(api_key_header) >= 8 else 'invalid'}"
)
if not verify_api_key(api_key_value):
suffix = api_key_value[-8:] if len(api_key_value) >= 8 else "invalid"
logger.warning("Invalid API key attempt: ...%s", suffix)
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid API key",
headers={"WWW-Authenticate": "ApiKey"},
)
return api_key_header
return api_key_value
async def verify_token(
@@ -153,12 +152,12 @@ async def verify_token(
payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM])
return payload
except JWTError as e:
logger.warning(f"Invalid JWT token: {str(e)}")
logger.warning("Invalid JWT token: %s", str(e))
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid authentication token",
headers={"WWW-Authenticate": "Bearer"},
)
) from e
async def require_admin_auth(
@@ -192,7 +191,7 @@ async def require_admin_auth(
)
return {"auth_type": "jwt", "payload": payload}
except JWTError as e:
logger.warning(f"Invalid JWT token: {str(e)}")
logger.warning("Invalid JWT token: %s", str(e))
# No valid authentication provided
raise HTTPException(
backend/app/main.py
+35 -31
View File
@@ -9,7 +9,7 @@ from app.core.security import add_api_key, generate_api_key, require_admin_auth
from app.middleware.security import SecurityHeadersMiddleware
from app.services.imap_client import IMAPClient
from app.services.report_store import ReportStore
from fastapi import BackgroundTasks, Depends, FastAPI, Request
from fastapi import Depends, FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import HTMLResponse
from fastapi.staticfiles import StaticFiles
@@ -27,7 +27,7 @@ last_check_time = None
async def scheduled_imap_polling():
"""Background task for periodically checking IMAP for new DMARC reports"""
global last_check_time
global last_check_time # pylint: disable=global-statement
try:
# How often to check for emails (in seconds)
@@ -46,19 +46,24 @@ async def scheduled_imap_polling():
if results["success"]:
logger.info(
f"IMAP polling completed: {results['processed']} emails processed, "
f"{results['reports_found']} reports found"
"IMAP polling completed: %s emails processed, %s reports found",
results["processed"],
results["reports_found"],
)
# If new domains were found, log them
if results["new_domains"]:
logger.info(f"New domains found: {', '.join(results['new_domains'])}")
logger.info(
"New domains found: %s", ", ".join(results["new_domains"])
)
else:
logger.error(f"IMAP polling failed: {results.get('error', 'Unknown error')}")
logger.error(
"IMAP polling failed: %s", results.get("error", "Unknown error")
)
except Exception as e:
logger.error(f"Error in IMAP polling task: {str(e)}")
except Exception as e: # pylint: disable=broad-exception-caught
logger.error("Error in IMAP polling task: %s", str(e))
# Wait for the next check interval
await asyncio.sleep(check_interval)
@@ -69,7 +74,7 @@ async def scheduled_imap_polling():
def create_app() -> FastAPI:
"""Create and configure the FastAPI application"""
app = FastAPI(
application = FastAPI(
title=settings.PROJECT_NAME,
openapi_url=f"{settings.API_V1_STR}/openapi.json",
version="0.1.0",
@@ -78,11 +83,11 @@ def create_app() -> FastAPI:
# Add security headers middleware
# Determine environment from settings or environment variable
environment = os.getenv("ENVIRONMENT", "development")
app.add_middleware(SecurityHeadersMiddleware, environment=environment)
application.add_middleware(SecurityHeadersMiddleware, environment=environment)
# Improved CORS configuration - restrict to specific methods and headers
if settings.BACKEND_CORS_ORIGINS:
app.add_middleware(
application.add_middleware(
CORSMiddleware,
allow_origins=[str(origin) for origin in settings.BACKEND_CORS_ORIGINS],
allow_credentials=True,
@@ -103,20 +108,20 @@ def create_app() -> FastAPI:
)
# Include API router
app.include_router(api_router, prefix=settings.API_V1_STR)
application.include_router(api_router, prefix=settings.API_V1_STR)
# Mount static files directory
app.mount(
application.mount(
"/static",
StaticFiles(directory=os.path.join(os.path.dirname(__file__), "static")),
name="static",
)
# Set up event handlers for startup and shutdown
@app.on_event("startup")
@application.on_event("startup")
async def startup_event():
"""Initialize background tasks and security on application startup"""
global background_task
global background_task # pylint: disable=global-statement
# Generate and provide admin API key
api_key = generate_api_key()
@@ -124,19 +129,20 @@ def create_app() -> FastAPI:
# Security: Log only last 8 characters for reference
logger.warning(
"=" * 80 + "\n"
"IMPORTANT: Admin API Key Generated\n"
f"API Key (last 8 chars): ...{api_key[-8:]}\n"
"%s\nIMPORTANT: Admin API Key Generated\n"
"API Key (last 8 chars): ...%s\n"
"Full key stored securely in memory.\n"
"For production, retrieve the key through secure configuration management.\n"
"Use this key in the X-API-Key header for admin endpoints.\n"
"=" * 80
"Use this key in the X-API-Key header for admin endpoints.\n%s",
"=" * 80,
api_key[-8:],
"=" * 80,
)
# In development, also log the full key for convenience
# This should be removed in production or controlled by environment variable
if os.getenv("ENVIRONMENT", "development") == "development":
logger.info(f"Development Mode - Full API Key: {api_key}")
logger.info("Development Mode - Full API Key: %s", api_key)
# Check if IMAP credentials are configured
if all([settings.IMAP_SERVER, settings.IMAP_USERNAME, settings.IMAP_PASSWORD]):
@@ -145,7 +151,7 @@ def create_app() -> FastAPI:
else:
logger.warning("IMAP credentials not fully configured, polling disabled")
@app.on_event("shutdown")
@application.on_event("shutdown")
async def shutdown_event():
"""Clean up background tasks on application shutdown"""
if background_task:
@@ -156,7 +162,7 @@ def create_app() -> FastAPI:
except asyncio.CancelledError:
pass
return app
return application
app = create_app()
@@ -202,9 +208,9 @@ async def domains(request: Request):
async def domain_details(request: Request, domain_id: str):
"""View detailed reports for a specific domain"""
store = ReportStore.get_instance()
domains = store.get_domains()
known_domains = store.get_domains()
if domain_id not in domains:
if domain_id not in known_domains:
# Domain not found, redirect to domains list
return templates.TemplateResponse(
"domains.html", {"request": request, "error": f"Domain {domain_id} not found"}
@@ -243,15 +249,13 @@ async def upload_page(request: Request):
# API endpoint to manually trigger IMAP polling
@app.post("/api/v1/admin/trigger-poll")
async def trigger_imap_poll(
background_tasks: BackgroundTasks, auth: dict = Depends(require_admin_auth)
):
async def trigger_imap_poll(auth: dict = Depends(require_admin_auth)):
"""
Manually trigger IMAP polling (admin only - requires authentication)
Security: Requires either X-API-Key header or Bearer token
"""
global last_check_time
global last_check_time # pylint: disable=global-statement
try:
# Create IMAP client and fetch reports
@@ -269,8 +273,8 @@ async def trigger_imap_poll(
"new_domains": results["new_domains"],
"authenticated_by": auth.get("auth_type"),
}
except Exception as e:
logger.error(f"Error triggering IMAP poll: {str(e)}")
except Exception as e: # pylint: disable=broad-exception-caught
logger.error("Error triggering IMAP poll: %s", str(e))
return {
"success": False,
"error": "Failed to trigger IMAP poll. Check server logs for details.",
backend/app/middleware/security.py
+7 -5
View File
@@ -78,11 +78,13 @@ class SecurityHeadersMiddleware(BaseHTTPMiddleware):
# See: https://developer.mozilla.org/en-US/docs/Web/HTTP/CSP
csp_directives = [
"default-src 'self'",
# TODO: Remove 'unsafe-inline' - requires moving inline scripts to external files
# TODO: Remove 'unsafe-eval' - no eval usage detected, safe to remove after testing
"script-src 'self' 'unsafe-inline' 'unsafe-eval' https://cdn.tailwindcss.com https://cdn.jsdelivr.net",
# TODO: Remove 'unsafe-inline' - requires moving inline styles to CSS or using nonces
"style-src 'self' 'unsafe-inline' https://fonts.googleapis.com https://cdn.jsdelivr.net",
# TODO: Remove 'unsafe-inline' - requires moving inline scripts to external files # pylint: disable=fixme
# TODO: Remove 'unsafe-eval' - no eval usage detected, safe to remove after testing # pylint: disable=fixme
"script-src 'self' 'unsafe-inline' 'unsafe-eval'"
" https://cdn.tailwindcss.com https://cdn.jsdelivr.net",
# TODO: Remove 'unsafe-inline' - requires moving inline styles to CSS or using nonces # pylint: disable=fixme
"style-src 'self' 'unsafe-inline' https://fonts.googleapis.com"
" https://cdn.jsdelivr.net",
"font-src 'self' https://fonts.gstatic.com",
"img-src 'self' data: https:",
"connect-src 'self'",
backend/app/services/dmarc_parser.py
+12 -8
View File
@@ -236,20 +236,24 @@ class DMARCParser:
# Log parse results for debugging
total_count = report["summary"]["total_count"]
logger.info(f"Parsed DMARC report for domain: {report.get('domain')}")
logger.info(f"Found {len(records)} record entries with {total_count} total messages")
logger.info("Parsed DMARC report for domain: %s", report.get("domain"))
logger.info(
f"Messages passed: {report['summary']['passed_count']}, "
f"failed: {report['summary']['failed_count']}"
"Found %s record entries with %s total messages", len(records), total_count
)
logger.info(
"Messages passed: %s, failed: %s",
report["summary"]["passed_count"],
report["summary"]["failed_count"],
)
if records:
logger.info(
f"Sample record - SPF: {records[0].get('spf_result')}, "
f"DKIM: {records[0].get('dkim_result')}"
"Sample record - SPF: %s, DKIM: %s",
records[0].get("spf_result"),
records[0].get("dkim_result"),
)
return report
except Exception as e:
logger.error(f"Error parsing DMARC XML: {str(e)}")
raise ValueError(f"Error parsing DMARC XML: {str(e)}")
logger.error("Error parsing DMARC XML: %s", str(e))
raise ValueError(f"Error parsing DMARC XML: {str(e)}") from e
backend/app/services/imap_client.py
+19 -17
View File
@@ -18,7 +18,7 @@ class IMAPClient:
Client for retrieving DMARC reports from an IMAP mailbox
"""
def __init__(
def __init__( # pylint: disable=too-many-positional-arguments,too-many-arguments
self,
server: str = None,
port: int = None,
@@ -63,7 +63,7 @@ class IMAPClient:
if mailbox_name.startswith(" "):
mailbox_name = mailbox_name[1:]
available_mailboxes.append(mailbox_name)
except Exception:
except Exception: # pylint: disable=broad-exception-caught
# Silently skip mailboxes that can't be parsed; they are simply
# omitted from the returned list so callers should expect it may
# be incomplete. Some IMAP servers return non-standard list
@@ -130,8 +130,8 @@ class IMAPClient:
}
return True, "Connection successful", stats
except Exception as e:
logger.error(f"IMAP connection test failed: {str(e)}")
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)}", {}
def _process_single_email(self, mail, email_id: bytes, stats: dict) -> None:
@@ -139,7 +139,7 @@ class IMAPClient:
try:
status, msg_data = mail.fetch(email_id, "(RFC822)")
if status != "OK":
logger.error(f"Error fetching email ID {email_id}")
logger.error("Error fetching email ID %s", email_id)
return
raw_email = msg_data[0][1]
@@ -155,7 +155,7 @@ class IMAPClient:
mail.store(email_id, "+FLAGS", "\\Deleted")
stats["processed"] += 1
except Exception as e:
except Exception as e: # pylint: disable=broad-exception-caught
error_msg = f"Error processing email ID {email_id}: {str(e)}"
logger.error(error_msg)
stats["errors"].append(error_msg)
@@ -225,8 +225,8 @@ class IMAPClient:
return stats
except Exception as e:
logger.error(f"Error fetching DMARC reports: {str(e)}")
except Exception as e: # pylint: disable=broad-exception-caught
logger.error("Error fetching DMARC reports: %s", str(e))
return {
"success": False,
"error": f"Error connecting to mailbox: {str(e)}",
@@ -339,12 +339,12 @@ class IMAPClient:
# Check content type
content_type = part.get_content_type()
if (
content_type == "application/zip"
or content_type == "application/gzip"
or content_type == "application/x-gzip"
or content_type == "application/xml"
or content_type == "text/xml"
if content_type in (
"application/zip",
"application/gzip",
"application/x-gzip",
"application/xml",
"text/xml",
):
return True
@@ -390,8 +390,10 @@ class IMAPClient:
self.report_store.add_report(report)
reports_found += 1
logger.info(f"Successfully processed DMARC report: {filename}")
except Exception as e:
logger.error(f"Error processing attachment {filename}: {str(e)}")
logger.info("Successfully processed DMARC report: %s", filename)
except Exception as e: # pylint: disable=broad-exception-caught
logger.error(
"Error processing attachment %s: %s", filename, str(e)
)
return reports_found
backend/app/services/report_store.py
+2 -2
View File
@@ -155,7 +155,7 @@ class ReportStore:
return sorted_reports[:limit]
return sorted_reports
def get_domain_sources(self, domain: str, days: int = 30) -> List[Dict[str, Any]]:
def get_domain_sources(self, domain: str, _days: int = 30) -> List[Dict[str, Any]]:
"""
Get sending sources for a domain
@@ -206,6 +206,6 @@ class ReportStore:
self.domain_summary.pop(domain, None)
self.domain_sources.pop(domain, None)
return True
except Exception:
except Exception: # pylint: disable=broad-exception-caught
# If any exception occurs during deletion, return False
return False
backend/app/tests/conftest.py
+5 -6
View File
@@ -1,9 +1,10 @@
# 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
import app.models.domain # 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
import pytest
from app.core.database import Base, get_db
from app.main import create_app
from app.services.report_store import ReportStore
from fastapi import FastAPI
from fastapi.testclient import TestClient
@@ -14,8 +15,6 @@ from sqlalchemy.orm import sessionmaker
@pytest.fixture()
def test_app() -> FastAPI:
"""Create a fresh FastAPI application instance for testing."""
from app.main import create_app
application = create_app()
return application
@@ -36,7 +35,7 @@ def db_session():
@pytest.fixture()
def client(test_app: FastAPI, db_session):
def client(test_app: FastAPI, db_session): # pylint: disable=redefined-outer-name
"""Create a TestClient with a DB override for the test app."""
def override_get_db():
backend/app/tests/test_data.py
+49
View File
@@ -0,0 +1,49 @@
"""Shared test data for DMARC report tests."""
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>
"""
backend/app/tests/test_dmarc_parser.py
+1 -48
View File
@@ -3,54 +3,7 @@ import zipfile
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>
"""
from app.tests.test_data import SAMPLE_XML
class TestDMARCParser:
backend/app/tests/test_reports_api.py
+1 -48
View File
@@ -1,56 +1,9 @@
import io
import zipfile
from app.tests.test_data import SAMPLE_XML
from fastapi.testclient import TestClient
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 _make_zip(xml_content: str) -> bytes:
"""Create a ZIP file containing the given XML content."""
backend/app/tests/test_security.py
+2 -3
View File
@@ -4,6 +4,7 @@ Security-focused tests for DMARQ application.
Covers API key management, domain validation, file upload limits, and XML parsing security.
"""
import app.services.dmarc_parser as parser_module
import pytest
from app.core.security import add_api_key, generate_api_key, verify_api_key
from app.services.dmarc_parser import DMARCParser
@@ -123,8 +124,6 @@ class TestXMLParsingSecurity:
"""Test XML parsing security (defusedxml, XXE protection)."""
def test_defusedxml_is_used(self):
import app.services.dmarc_parser as parser_module
assert hasattr(parser_module, "ET")
module_info = str(getattr(parser_module.ET, "__name__", "")) + str(
getattr(parser_module.ET, "__module__", "")
@@ -149,5 +148,5 @@ class TestXMLParsingSecurity:
result = DMARCParser.parse_file(xxe_payload, "test.xml")
org_name = result.get("org_name", "")
assert "root:" not in org_name and "/bin" not in org_name
except Exception:
except Exception: # pylint: disable=broad-exception-caught
pass # Expected – defusedxml blocks DTD processing
backend/app/utils/domain_validator.py
+2 -3
View File
@@ -56,7 +56,7 @@ def _validate_domain_labels(
return True, None, None
def validate_domain(
def validate_domain( # pylint: disable=too-many-return-statements
domain_name: str, check_dns: bool = True
) -> Tuple[bool, Optional[str], Optional[str]]:
"""
@@ -103,7 +103,6 @@ def validate_domain(
if check_dns:
try:
socket.gethostbyname(domain_name)
return True, None, None
except socket.gaierror:
# We could consider this valid if we don't require DNS resolution,
# but since DMARC requires valid DNS, we'll mark it as warning
@@ -133,7 +132,7 @@ def validate_domain_config(domain_data: Dict) -> Dict[str, Union[bool, str]]:
# Validate domain name
if "name" in domain_data:
# Don't check DNS for domain config validation
is_valid, error_msg, error_code = validate_domain(domain_data["name"], check_dns=False)
is_valid, error_msg, _ = validate_domain(domain_data["name"], check_dns=False)
if not is_valid:
errors["name"] = error_msg
else:
backend/app/utils/stats_summarizer.py
+12 -11
View File
@@ -63,10 +63,10 @@ class StatsSummarizer:
return None
# Read cache file
with open(cache_file, "r") as f:
with open(cache_file, "r", encoding="utf-8") as f:
return json.load(f)
except Exception as e:
logger.warning(f"Error reading cache file {cache_file}: {str(e)}")
except Exception as e: # pylint: disable=broad-exception-caught
logger.warning("Error reading cache file %s: %s", cache_file, str(e))
return None
def save_summary(self, stats: Dict[str, Any], domain_id: Optional[str] = None) -> bool:
@@ -87,12 +87,12 @@ class StatsSummarizer:
stats["cached_at"] = datetime.now().isoformat()
# Write to cache file
with open(cache_file, "w") as f:
with open(cache_file, "w", encoding="utf-8") as f:
json.dump(stats, f)
return True
except Exception as e:
logger.error(f"Error writing cache file {cache_file}: {str(e)}")
except Exception as e: # pylint: disable=broad-exception-caught
logger.error("Error writing cache file %s: %s", cache_file, str(e))
return False
def invalidate_cache(self, domain_id: Optional[str] = None) -> None:
@@ -126,12 +126,13 @@ class StatsSummarizer:
"""
if domain_id is None:
return os.path.join(self.cache_dir, "global_summary.json")
else:
# Sanitize domain_id to use as filename
safe_domain = domain_id.replace(".", "_").replace("/", "_")
return os.path.join(self.cache_dir, f"domain_{safe_domain}.json")
# Sanitize domain_id to use as filename
safe_domain = domain_id.replace(".", "_").replace("/", "_")
return os.path.join(self.cache_dir, f"domain_{safe_domain}.json")
def calculate_summary_statistics(self, db, domain_id: Optional[str] = None) -> Dict[str, Any]:
def calculate_summary_statistics(
self, _db, domain_id: Optional[str] = None
) -> Dict[str, Any]:
"""
Calculate summary statistics from the database