From 50aa5bd5daf7bc76a3b61479f6c4c1d4cc4912b7 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sun, 29 Mar 2026 11:21:35 +0000 Subject: [PATCH] Fix pylint warnings: logging, globals, exceptions, imports, duplicates Agent-Logs-Url: https://github.com/christianlouis/dmarq/sessions/576427d4-4f6a-46d2-b75f-6862ecbcf526 Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com> --- backend/app/api/api_v1/endpoints/domains.py | 8 ++- backend/app/api/api_v1/endpoints/reports.py | 24 ++++---- backend/app/core/config.py | 13 ++-- backend/app/core/security.py | 21 ++++--- backend/app/main.py | 66 +++++++++++---------- backend/app/middleware/security.py | 12 ++-- backend/app/services/dmarc_parser.py | 20 ++++--- backend/app/services/imap_client.py | 36 +++++------ backend/app/services/report_store.py | 4 +- backend/app/tests/conftest.py | 11 ++-- backend/app/tests/test_data.py | 49 +++++++++++++++ backend/app/tests/test_dmarc_parser.py | 49 +-------------- backend/app/tests/test_reports_api.py | 49 +-------------- backend/app/tests/test_security.py | 5 +- backend/app/utils/domain_validator.py | 5 +- backend/app/utils/stats_summarizer.py | 23 +++---- 16 files changed, 182 insertions(+), 213 deletions(-) create mode 100644 backend/app/tests/test_data.py diff --git a/backend/app/api/api_v1/endpoints/domains.py b/backend/app/api/api_v1/endpoints/domains.py index 451019a..d897871 100644 --- a/backend/app/api/api_v1/endpoints/domains.py +++ b/backend/app/api/api_v1/endpoints/domains.py @@ -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 diff --git a/backend/app/api/api_v1/endpoints/reports.py b/backend/app/api/api_v1/endpoints/reports.py index 419c6c1..39b9ff0 100644 --- a/backend/app/api/api_v1/endpoints/reports.py +++ b/backend/app/api/api_v1/endpoints/reports.py @@ -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) diff --git a/backend/app/core/config.py b/backend/app/core/config.py index a25d929..97c0003 100644 --- a/backend/app/core/config.py +++ b/backend/app/core/config.py @@ -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) diff --git a/backend/app/core/security.py b/backend/app/core/security.py index ccd4be2..56836ca 100644 --- a/backend/app/core/security.py +++ b/backend/app/core/security.py @@ -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( diff --git a/backend/app/main.py b/backend/app/main.py index fdb50a8..dba8d6c 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -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.", diff --git a/backend/app/middleware/security.py b/backend/app/middleware/security.py index aba4c3a..6bb9700 100644 --- a/backend/app/middleware/security.py +++ b/backend/app/middleware/security.py @@ -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'", diff --git a/backend/app/services/dmarc_parser.py b/backend/app/services/dmarc_parser.py index c5944f2..2adaf61 100644 --- a/backend/app/services/dmarc_parser.py +++ b/backend/app/services/dmarc_parser.py @@ -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 diff --git a/backend/app/services/imap_client.py b/backend/app/services/imap_client.py index 7813ed9..73780ed 100644 --- a/backend/app/services/imap_client.py +++ b/backend/app/services/imap_client.py @@ -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 diff --git a/backend/app/services/report_store.py b/backend/app/services/report_store.py index a605a29..d62e220 100644 --- a/backend/app/services/report_store.py +++ b/backend/app/services/report_store.py @@ -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 diff --git a/backend/app/tests/conftest.py b/backend/app/tests/conftest.py index 97c2ac5..61088df 100644 --- a/backend/app/tests/conftest.py +++ b/backend/app/tests/conftest.py @@ -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(): diff --git a/backend/app/tests/test_data.py b/backend/app/tests/test_data.py new file mode 100644 index 0000000..24642ec --- /dev/null +++ b/backend/app/tests/test_data.py @@ -0,0 +1,49 @@ +"""Shared test data for DMARC report tests.""" + +SAMPLE_XML = """\ + + + + google.com + noreply-dmarc-support@google.com + 123456789 + + 1597449600 + 1597535999 + + + + example.com + r + r +

none

+ none + 100 + + + + 203.0.113.1 + 2 + + none + pass + fail + + + + example.com + + + + example.com + pass + default + + + example.com + fail + + + + +""" diff --git a/backend/app/tests/test_dmarc_parser.py b/backend/app/tests/test_dmarc_parser.py index 9a8adc7..72a3934 100644 --- a/backend/app/tests/test_dmarc_parser.py +++ b/backend/app/tests/test_dmarc_parser.py @@ -3,54 +3,7 @@ import zipfile import pytest from app.services.dmarc_parser import DMARCParser - -SAMPLE_XML = """\ - - - - google.com - noreply-dmarc-support@google.com - 123456789 - - 1597449600 - 1597535999 - - - - example.com - r - r -

none

- none - 100 - - - - 203.0.113.1 - 2 - - none - pass - fail - - - - example.com - - - - example.com - pass - default - - - example.com - fail - - - - -""" +from app.tests.test_data import SAMPLE_XML class TestDMARCParser: diff --git a/backend/app/tests/test_reports_api.py b/backend/app/tests/test_reports_api.py index 337b712..f7a3a3a 100644 --- a/backend/app/tests/test_reports_api.py +++ b/backend/app/tests/test_reports_api.py @@ -1,56 +1,9 @@ import io import zipfile +from app.tests.test_data import SAMPLE_XML from fastapi.testclient import TestClient -SAMPLE_XML = """\ - - - - google.com - noreply-dmarc-support@google.com - 123456789 - - 1597449600 - 1597535999 - - - - example.com - r - r -

none

- none - 100 - - - - 203.0.113.1 - 2 - - none - pass - fail - - - - example.com - - - - example.com - pass - default - - - example.com - fail - - - - -""" - def _make_zip(xml_content: str) -> bytes: """Create a ZIP file containing the given XML content.""" diff --git a/backend/app/tests/test_security.py b/backend/app/tests/test_security.py index 466a136..d99bebe 100644 --- a/backend/app/tests/test_security.py +++ b/backend/app/tests/test_security.py @@ -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 diff --git a/backend/app/utils/domain_validator.py b/backend/app/utils/domain_validator.py index a329249..faca6c8 100644 --- a/backend/app/utils/domain_validator.py +++ b/backend/app/utils/domain_validator.py @@ -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: diff --git a/backend/app/utils/stats_summarizer.py b/backend/app/utils/stats_summarizer.py index 1cfe071..098f6f5 100644 --- a/backend/app/utils/stats_summarizer.py +++ b/backend/app/utils/stats_summarizer.py @@ -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