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

Add comprehensive documentation for DMARQ, including user guides, deployment instructions, and feature descriptions

Browse Source 
- Created main documentation index and user guide with sections on getting started, dashboard overview, managing domains, and reports.
- Added detailed deployment guide for Docker and manual installation.
- Included user-friendly explanations of DMARC, its benefits, and how to manage domains and reports.
- Implemented visual assets for dashboard, domains, IMAP, and reports.
- Established requirements for documentation build using MkDocs and Material theme.
- Integrated navigation structure for easy access to all documentation sections.
This commit is contained in:
Christian Krakau-Louis
2025-04-21 01:49:34 +02:00
parent 1b79ec4f20
commit 5e8b1f033f
29 changed files with 1947 additions and 110 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/api/api_v1/endpoints/domains.py
+79 -1
View File
@@ -315,4 +315,82 @@ async def get_domain_sources(
return DomainSourcesResponse(
sources=source_entries
)
)
@router.delete("/{domain_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_domain(domain_id: str = Path(..., title="The domain ID or name")):
"""
Delete a domain and all associated data.
This performs a full cleanup of all reports and records related to this domain.
"""
store = ReportStore.get_instance()
domains = store.get_domains()
if domain_id not in domains:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Domain not found",
)
# Perform deletion with cleanup
deleted = store.delete_domain_with_cleanup(domain_id)
if not deleted:
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="Failed to delete domain",
)
# Return 204 No Content on success
return None
@router.get("/search", response_model=List[DomainResponse])
async def search_domains(
q: Optional[str] = Query(None, title="Search query for domain name or description"),
policy: Optional[str] = Query(None, title="Filter by DMARC policy"),
page: int = Query(1, title="Page number", ge=1),
limit: int = Query(10, title="Number of domains per page", ge=1, le=100)
):
"""
Search domains with filtering and pagination.
This supports searching by domain name/description and filtering by DMARC policy.
Args:
q: Optional search query for domain name or description
policy: Optional filter by DMARC policy (none, quarantine, reject)
page: Page number (1-based)
limit: Number of domains per page (max 100)
"""
store = ReportStore.get_instance()
domains = store.get_domains()
summaries = store.get_all_domain_summaries()
# Apply search filter if provided
filtered_domains = []
for domain_name in domains:
summary = summaries.get(domain_name, {})
# Skip domain if it doesn't match the search query
if q and q.lower() not in domain_name.lower():
continue
# Skip domain if it doesn't match the policy filter
if policy and summary.get("policy") != policy:
continue
# Domain passed all filters
filtered_domains.append({
"name": domain_name,
"description": "", # No description in in-memory store
"policy": summary.get("policy", "unknown"),
"reports_count": summary.get("reports_processed", 0),
"emails_count": summary.get("total_count", 0),
"compliance_rate": summary.get("compliance_rate", 0.0)
})
# Apply pagination
start_idx = (page - 1) * limit
end_idx = start_idx + limit
paginated_domains = filtered_domains[start_idx:end_idx]
return [DomainResponse(**domain) for domain in paginated_domains]
backend/app/api/api_v1/endpoints/imap.py
+59 -3
View File
@@ -11,10 +11,11 @@ async def test_imap_connection(
server: str = None,
port: int = 993,
username: str = None,
password: str = None
password: str = None,
ssl: bool = True
) -> Dict[str, Any]:
"""
Test connection to an IMAP server
Test connection to an IMAP server and gather mailbox statistics
"""
imap_client = IMAPClient(
server=server,
@@ -23,11 +24,15 @@ async def test_imap_connection(
password=password
)
success, message = imap_client.test_connection()
success, message, stats = imap_client.test_connection()
return {
"success": success,
"message": message,
"message_count": stats.get("message_count", 0),
"unread_count": stats.get("unread_count", 0),
"dmarc_count": stats.get("dmarc_count", 0),
"available_mailboxes": stats.get("available_mailboxes", []),
"timestamp": datetime.now().isoformat()
}
@@ -62,4 +67,55 @@ async def fetch_imap_reports(
"new_domains": results["new_domains"],
"errors": results["errors"] if "errors" in results and results["errors"] else None,
"timestamp": datetime.now().isoformat()
}
@router.get("/status")
async def get_imap_status() -> Dict[str, Any]:
"""
Get the current status of IMAP polling background processes
"""
# In a real implementation this would check a persistent store
# or a global variable tracking the status of background tasks
# For now, returning mock data as this is MVP
# Get the last check time if available
last_check_time = None
try:
# In a production app, this would be stored in database
# For MVP, using a simple file-based approach
import os
status_file = os.path.join(os.path.dirname(__file__), "../../../../../tmp/imap_last_check.txt")
if os.path.exists(status_file):
with open(status_file, "r") as f:
last_check_time = f.read().strip()
except:
pass
# If status file doesn't exist, create the directory
try:
os.makedirs(os.path.dirname(os.path.join(os.path.dirname(__file__), "../../../../../tmp")), exist_ok=True)
except:
pass
# For demonstration purposes, update the last check time to now
# In a real app, this would be updated by the background process
try:
with open(os.path.join(os.path.dirname(__file__), "../../../../../tmp/imap_last_check.txt"), "w") as f:
now = datetime.now().isoformat()
f.write(now)
# If there was no previous check time, set it to now
if not last_check_time:
last_check_time = now
except:
pass
# Return the status
return {
"is_running": True, # In a real app, check if the background task is running
"last_check": last_check_time,
"next_check": None, # In production, this would be calculated based on polling interval
"messages_processed": 0, # In production, this would track actual messages processed
"reports_found": 0, # In production, this would track reports found
"timestamp": datetime.now().isoformat()
}
backend/app/api/api_v1/endpoints/reports.py
+80 -1
View File
@@ -33,6 +33,14 @@ class ReportSummary(BaseModel):
passed_count: int
failed_count: int
class PaginatedReportResponse(BaseModel):
"""Paginated reports response model"""
total: int
page: int
page_size: int
total_pages: int
reports: List[ReportSummary]
@router.post("/upload", response_model=UploadResponse)
async def upload_report(file: UploadFile = File(...)):
"""
@@ -132,4 +140,75 @@ async def get_domain_reports(domain: str):
failed_count=report.get("summary", {}).get("failed_count", 0)
)
for report in reports
]
]
@router.get("/domain/{domain}/reports/paginated", response_model=PaginatedReportResponse)
async def get_domain_reports_paginated(
domain: str,
page: int = 1,
page_size: int = 10,
sort_by: str = "end_date",
sort_order: str = "desc"
):
"""
Get paginated reports for a specific domain with sorting options
Args:
domain: Domain name
page: Page number (1-based)
page_size: Number of reports per page
sort_by: Field to sort by (report_id, org_name, begin_date, end_date, total_count)
sort_order: Sort order (asc or desc)
"""
store = ReportStore.get_instance()
all_reports = store.get_domain_reports(domain)
if not all_reports:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"No reports found for domain {domain}"
)
# Apply sorting
valid_sort_fields = ["report_id", "org_name", "begin_date", "end_date", "total_count"]
sort_field = sort_by if sort_by in valid_sort_fields else "end_date"
if sort_field == "total_count":
all_reports.sort(
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")
)
# Apply pagination
total = len(all_reports)
total_pages = (total + page_size - 1) // page_size
start_idx = (page - 1) * page_size
end_idx = start_idx + page_size
paginated_reports = all_reports[start_idx:end_idx]
# Format reports
report_entries = [
ReportSummary(
report_id=report.get("report_id", ""),
org_name=report.get("org_name", ""),
begin_date=report.get("begin_date", ""),
end_date=report.get("end_date", ""),
total_count=report.get("summary", {}).get("total_count", 0),
passed_count=report.get("summary", {}).get("passed_count", 0),
failed_count=report.get("summary", {}).get("failed_count", 0)
)
for report in paginated_reports
]
return PaginatedReportResponse(
total=total,
page=page,
page_size=page_size,
total_pages=total_pages,
reports=report_entries
)
backend/app/api/api_v1/endpoints/stats.py
+75
View File
@@ -0,0 +1,75 @@
from typing import Dict, Any, List, Optional
from fastapi import APIRouter, Depends, Query, Path
from sqlalchemy.orm import Session
from app.core.database import get_db
from app.utils.stats_summarizer import StatsSummarizer
router = APIRouter()
@router.get("/dashboard")
async def get_dashboard_statistics(
db: Session = Depends(get_db),
force_refresh: bool = Query(False, title="Force refresh of statistics"),
period_days: int = Query(30, title="Period in days for time-based statistics")
) -> Dict[str, Any]:
"""
Get optimized statistics for the dashboard using cached data when possible.
This endpoint provides efficient access to statistics for large datasets.
Args:
force_refresh: If True, invalidate cache and recalculate statistics
period_days: Period in days for time-based statistics (default: 30)
Returns:
Dictionary with dashboard statistics
"""
# Initialize statistics summarizer
stats_summarizer = StatsSummarizer()
# If force refresh, invalidate cache
if force_refresh:
stats_summarizer.invalidate_cache()
# Get statistics (from cache or calculate if needed)
stats = stats_summarizer.calculate_summary_statistics(db)
# Add version and timestamp
stats["api_version"] = "1.0"
stats["period_days"] = period_days
return stats
@router.get("/domain/{domain_id}")
async def get_domain_statistics(
domain_id: str = Path(..., title="The domain ID or name"),
db: Session = Depends(get_db),
force_refresh: bool = Query(False, title="Force refresh of statistics"),
period_days: int = Query(30, title="Period in days for time-based statistics")
) -> Dict[str, Any]:
"""
Get optimized statistics for a specific domain using cached data when possible.
Args:
domain_id: The domain ID or name
force_refresh: If True, invalidate cache and recalculate statistics
period_days: Period in days for time-based statistics (default: 30)
Returns:
Dictionary with domain statistics
"""
# Initialize statistics summarizer
stats_summarizer = StatsSummarizer()
# If force refresh, invalidate domain cache
if force_refresh:
stats_summarizer.invalidate_cache(domain_id)
# Get domain statistics (from cache or calculate if needed)
stats = stats_summarizer.calculate_summary_statistics(db, domain_id)
# Add version and timestamp
stats["api_version"] = "1.0"
stats["period_days"] = period_days
return stats