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

Fix code formatting and linting issues

Browse Source 
- Auto-format all Python files with black and isort
- Remove unused imports with autoflake
- Fix flake8 issues (missing newlines, blank lines, etc.)
- Fix nonlocal/global scope issues in main.py
- Fix security.py import order (E402)
- Remove f-string without placeholders
- Add nosec comment for intentional exception handling
- Fix test imports to match refactored DMARCParser API

Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>
This commit is contained in:
copilot-swe-agent[bot]
2026-02-09 12:08:51 +00:00
parent f6908fe9ec
commit 6ae017b142
28 changed files with 999 additions and 956 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/utils/__init__.py
+1 -1
View File
@@ -1,3 +1,3 @@
"""
Utilities for DMARQ application.
"""
"""
backend/app/utils/domain_validator.py
+52 -31
View File
@@ -1,11 +1,13 @@
import html
import re
import socket
import html
from typing import Dict, Tuple, Union, Optional
from typing import Dict, Optional, Tuple, Union
# Error codes for structured error handling
class DomainValidationError:
"""Domain validation error codes"""
EMPTY = "empty"
TOO_LONG = "too_long"
INVALID_FORMAT = "invalid_format"
@@ -15,14 +17,16 @@ class DomainValidationError:
DNS_RESOLUTION_FAILED = "dns_resolution_failed"
def validate_domain(domain_name: str, check_dns: bool = True) -> Tuple[bool, Optional[str], Optional[str]]:
def validate_domain(
domain_name: str, check_dns: bool = True
) -> Tuple[bool, Optional[str], Optional[str]]:
"""
Validates a domain name for format and optionally resolvability.
Args:
domain_name: The domain name to validate
check_dns: Whether to perform DNS resolution check (default: True)
Returns:
Tuple containing (is_valid, error_message, error_code)
- is_valid: Boolean indicating if domain is valid
@@ -32,35 +36,51 @@ def validate_domain(domain_name: str, check_dns: bool = True) -> Tuple[bool, Opt
# Security: Check for empty or None domain
if not domain_name:
return False, "Domain name cannot be empty", DomainValidationError.EMPTY
# Security: Check maximum length (DNS standard is 253 characters)
if len(domain_name) > 253:
return False, "Domain name too long (max 253 characters)", DomainValidationError.TOO_LONG
# Security: Check for whitespace
if ' ' in domain_name or '\t' in domain_name or '\n' in domain_name:
return False, "Domain name cannot contain whitespace", DomainValidationError.INVALID_CHARACTERS
if " " in domain_name or "\t" in domain_name or "\n" in domain_name:
return (
False,
"Domain name cannot contain whitespace",
DomainValidationError.INVALID_CHARACTERS,
)
# Security: Check for suspicious characters
if any(char in domain_name for char in ['<', '>', '"', "'", '\\', '|', ';', '&', '$', '`']):
return False, "Domain name contains invalid characters", DomainValidationError.INVALID_CHARACTERS
if any(char in domain_name for char in ["<", ">", '"', "'", "\\", "|", ";", "&", "$", "`"]):
return (
False,
"Domain name contains invalid characters",
DomainValidationError.INVALID_CHARACTERS,
)
# Check domain format with regex
# This regex allows domain names with alphanumeric characters, hyphens,
# and periods as separators. It enforces proper domain structure.
# Updated to be more strict and prevent potential attacks
domain_pattern = r'^(?:[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?\.)+[a-z0-9][a-z0-9-]{0,61}[a-z0-9]$'
domain_pattern = r"^(?:[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?\.)+[a-z0-9][a-z0-9-]{0,61}[a-z0-9]$"
if not re.match(domain_pattern, domain_name.lower()):
return False, "Invalid domain format", DomainValidationError.INVALID_FORMAT
# Security: Check each label length (max 63 characters per label)
labels = domain_name.split('.')
labels = domain_name.split(".")
for label in labels:
if len(label) > 63:
return False, f"Domain label too long: '{label}' (max 63 characters per label)", DomainValidationError.LABEL_TOO_LONG
if label.startswith('-') or label.endswith('-'):
return False, f"Domain label cannot start or end with hyphen: '{label}'", DomainValidationError.INVALID_LABEL
return (
False,
f"Domain label too long: '{label}' (max 63 characters per label)",
DomainValidationError.LABEL_TOO_LONG,
)
if label.startswith("-") or label.endswith("-"):
return (
False,
f"Domain label cannot start or end with hyphen: '{label}'",
DomainValidationError.INVALID_LABEL,
)
# Check if domain exists by attempting to resolve DNS (optional)
if check_dns:
try:
@@ -69,25 +89,29 @@ def validate_domain(domain_name: str, check_dns: bool = True) -> Tuple[bool, Opt
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
return False, "Domain could not be resolved (DNS lookup failed)", DomainValidationError.DNS_RESOLUTION_FAILED
return (
False,
"Domain could not be resolved (DNS lookup failed)",
DomainValidationError.DNS_RESOLUTION_FAILED,
)
return True, None, None
def validate_domain_config(domain_data: Dict) -> Dict[str, Union[bool, str]]:
"""
Validates domain configuration data for creating or updating domains.
Args:
domain_data: Dictionary with domain configuration
Returns:
Dictionary with validation results containing:
- valid: Boolean indicating if configuration is valid
- errors: Dict of field-specific errors
"""
errors = {}
# Validate domain name
if "name" in domain_data:
# Don't check DNS for domain config validation
@@ -96,7 +120,7 @@ def validate_domain_config(domain_data: Dict) -> Dict[str, Union[bool, str]]:
errors["name"] = error_msg
else:
errors["name"] = "Domain name is required"
# Validate description (optional but with max length)
if "description" in domain_data and domain_data["description"]:
if len(domain_data["description"]) > 500:
@@ -105,9 +129,6 @@ def validate_domain_config(domain_data: Dict) -> Dict[str, Union[bool, str]]:
escaped = html.escape(domain_data["description"])
if escaped != domain_data["description"]:
errors["description"] = "Description contains potentially unsafe HTML content"
# Return validation results
return {
"valid": len(errors) == 0,
"errors": errors
}
return {"valid": len(errors) == 0, "errors": errors}
backend/app/utils/stats_summarizer.py
+50 -43
View File
@@ -1,97 +1,104 @@
from datetime import datetime, timedelta
from typing import Dict, List, Any, Optional
import logging
import json
import logging
import os
from datetime import datetime, timedelta
from typing import Any, Dict, Optional
# Setup logger
logger = logging.getLogger(__name__)
class StatsSummarizer:
"""
Utility class for summarizing and caching dashboard statistics
to improve performance with large datasets.
"""
def __init__(self, cache_dir: str = None):
"""
Initialize the stats summarizer with optional cache directory
Args:
cache_dir: Directory to store cached statistics (defaults to tmp/stats)
"""
if cache_dir is None:
# Default cache directory is tmp/stats under the project root
self.cache_dir = os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(__file__)))), "tmp", "stats")
self.cache_dir = os.path.join(
os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(__file__)))),
"tmp",
"stats",
)
else:
self.cache_dir = cache_dir
# Create cache directory if it doesn't exist
os.makedirs(self.cache_dir, exist_ok=True)
def get_cached_summary(self, domain_id: Optional[str] = None, max_age_minutes: int = 60) -> Optional[Dict[str, Any]]:
def get_cached_summary(
self, domain_id: Optional[str] = None, max_age_minutes: int = 60
) -> Optional[Dict[str, Any]]:
"""
Get cached summary statistics if available and not too old
Args:
domain_id: Optional domain ID to get domain-specific stats
If None, gets global summary
max_age_minutes: Maximum age of cache in minutes
Returns:
Cached statistics or None if not available or too old
"""
cache_file = self._get_cache_filename(domain_id)
try:
if not os.path.exists(cache_file):
return None
# Check file modification time
mtime = os.path.getmtime(cache_file)
file_age = datetime.now() - datetime.fromtimestamp(mtime)
# If cache is too old, return None
if file_age > timedelta(minutes=max_age_minutes):
return None
# Read cache file
with open(cache_file, 'r') as f:
with open(cache_file, "r") as f:
return json.load(f)
except Exception as e:
logger.warning(f"Error reading cache file {cache_file}: {str(e)}")
return None
def save_summary(self, stats: Dict[str, Any], domain_id: Optional[str] = None) -> bool:
"""
Save summary statistics to cache
Args:
stats: Dictionary of statistics to cache
domain_id: Optional domain ID for domain-specific stats
Returns:
True if save was successful, False otherwise
"""
cache_file = self._get_cache_filename(domain_id)
try:
# Add timestamp
stats["cached_at"] = datetime.now().isoformat()
# Write to cache file
with open(cache_file, 'w') as f:
with open(cache_file, "w") as f:
json.dump(stats, f)
return True
except Exception as e:
logger.error(f"Error writing cache file {cache_file}: {str(e)}")
return False
def invalidate_cache(self, domain_id: Optional[str] = None) -> None:
"""
Invalidate cache for a domain or all domains
Args:
domain_id: Optional domain ID to invalidate specific domain cache
If None, invalidates global summary cache
@@ -106,14 +113,14 @@ class StatsSummarizer:
cache_file = self._get_cache_filename(domain_id)
if os.path.exists(cache_file):
os.remove(cache_file)
def _get_cache_filename(self, domain_id: Optional[str] = None) -> str:
"""
Get the filename for a cache file
Args:
domain_id: Optional domain ID for domain-specific cache
Returns:
Path to the cache file
"""
@@ -123,31 +130,31 @@ class StatsSummarizer:
# 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]:
"""
Calculate summary statistics from the database
Args:
db: Database session
domain_id: Optional domain ID to calculate domain-specific stats
Returns:
Dictionary with summary statistics
"""
# In a real implementation, this would query the database
# using SQLAlchemy models and calculate statistics
# For now, we'll return mock statistics
# First check if we have cached stats
cached_stats = self.get_cached_summary(domain_id)
if cached_stats:
return cached_stats
# If no cached stats, calculate from database
# In a real implementation, this would be done with SQL queries
# optimized for performance with large datasets
# For now, mock statistics
if domain_id is None:
# Global statistics
@@ -160,7 +167,7 @@ class StatsSummarizer:
"top_sources": [
{"ip": "192.168.1.1", "count": 150},
{"ip": "10.0.0.1", "count": 120},
{"ip": "172.16.0.1", "count": 100}
{"ip": "172.16.0.1", "count": 100},
],
"compliance_trend": [
{"date": "2025-04-13", "rate": 85.5},
@@ -169,8 +176,8 @@ class StatsSummarizer:
{"date": "2025-04-16", "rate": 87.3},
{"date": "2025-04-17", "rate": 87.9},
{"date": "2025-04-18", "rate": 88.4},
{"date": "2025-04-19", "rate": 88.0}
]
{"date": "2025-04-19", "rate": 88.0},
],
}
else:
# Domain-specific statistics
@@ -183,7 +190,7 @@ class StatsSummarizer:
"sources": [
{"ip": "192.168.1.1", "count": 100, "spf": "pass", "dkim": "pass"},
{"ip": "10.0.0.1", "count": 80, "spf": "pass", "dkim": "fail"},
{"ip": "172.16.0.1", "count": 70, "spf": "fail", "dkim": "pass"}
{"ip": "172.16.0.1", "count": 70, "spf": "fail", "dkim": "pass"},
],
"compliance_trend": [
{"date": "2025-04-13", "rate": 85.0},
@@ -192,11 +199,11 @@ class StatsSummarizer:
{"date": "2025-04-16", "rate": 87.5},
{"date": "2025-04-17", "rate": 88.0},
{"date": "2025-04-18", "rate": 88.5},
{"date": "2025-04-19", "rate": 88.0}
]
{"date": "2025-04-19", "rate": 88.0},
],
}
# Cache the statistics
self.save_summary(stats, domain_id)
return stats
return stats