Domain Audit Use Case¶

# 1. Collect client domains for assessment
client_domains = [
    "client-company.com",
    "marketing.client-company.com", 
    "support.client-company.com",
    "mail.client-company.com"
]

# 2. Initialize audit session
audit_session = {
    "client_id": "client_123",
    "audit_date": datetime.now(),
    "domains": client_domains,
    "scope": "comprehensive",  # basic, comprehensive, compliance
    "standards": ["SOC2", "ISO27001"]
}

import reputeapi
from datetime import datetime
import json

# Initialize ReputeAPI client
client = reputeapi.Client(api_key="your_api_key")

def conduct_bulk_audit(domains):
    """Perform comprehensive audit of multiple domains"""
    results = []

    for domain in domains:
        try:
            # Get comprehensive security analysis
            validation = client.check(domain)

            # Calculate risk metrics
            risk_assessment = calculate_risk_score(validation)

            # Generate recommendations
            recommendations = generate_recommendations(validation)

            audit_result = {
                "domain": domain,
                "timestamp": datetime.now().isoformat(),
                "security_score": validation.score,
                "risk_level": risk_assessment["level"],
                "findings": validation.findings,
                "recommendations": recommendations,
                "compliance_status": check_compliance(validation)
            }

            results.append(audit_result)

        except Exception as e:
            print(f"Error auditing {domain}: {e}")
            continue

    return results

# Execute audit
audit_results = conduct_bulk_audit(client_domains)

def calculate_risk_score(validation):
    """Calculate comprehensive risk score based on findings"""

    risk_factors = {
        "spf_missing": 30,
        "dkim_missing": 25, 
        "dmarc_missing": 35,
        "weak_spf_policy": 15,
        "weak_dmarc_policy": 20,
        "dns_vulnerabilities": 25,
        "configuration_errors": 10
    }

    total_risk = 0
    findings = []

    # Analyze SPF configuration
    if not validation.spf.valid:
        total_risk += risk_factors["spf_missing"]
        findings.append({
            "type": "SPF Missing",
            "severity": "High",
            "description": "No SPF record found",
            "impact": "Email spoofing vulnerability"
        })
    elif validation.spf.policy == "~all":  # SoftFail
        total_risk += risk_factors["weak_spf_policy"]
        findings.append({
            "type": "Weak SPF Policy", 
            "severity": "Medium",
            "description": "SPF uses soft fail (~all)",
            "impact": "Reduced protection against spoofing"
        })

    # Analyze DKIM configuration
    if not validation.dkim.valid:
        total_risk += risk_factors["dkim_missing"]
        findings.append({
            "type": "DKIM Missing",
            "severity": "High", 
            "description": "No valid DKIM signature found",
            "impact": "Message authenticity cannot be verified"
        })

    # Analyze DMARC configuration
    if not validation.dmarc.valid:
        total_risk += risk_factors["dmarc_missing"]
        findings.append({
            "type": "DMARC Missing",
            "severity": "Critical",
            "description": "No DMARC policy found",
            "impact": "No protection against email spoofing"
        })
    elif validation.dmarc.policy in ["none", "quarantine"]:
        total_risk += risk_factors["weak_dmarc_policy"]
        findings.append({
            "type": "Weak DMARC Policy",
            "severity": "Medium",
            "description": f"DMARC policy is '{validation.dmarc.policy}'",
            "impact": "Limited protection against spoofing"
        })

    # Determine risk level
    if total_risk >= 80:
        risk_level = "Critical"
    elif total_risk >= 60:
        risk_level = "High"
    elif total_risk >= 30:
        risk_level = "Medium"
    else:
        risk_level = "Low"

    return {
        "score": total_risk,
        "level": risk_level,
        "findings": findings,
        "max_possible": 100
    }

def check_compliance(validation):
    """Map findings to compliance standards"""

    compliance_mapping = {
        "SOC2": {
            "CC6.1": check_access_controls(validation),
            "CC6.7": check_transmission_integrity(validation), 
            "CC7.1": check_system_boundaries(validation)
        },
        "ISO27001": {
            "A.13.2.1": check_information_transfer(validation),
            "A.13.2.3": check_electronic_messaging(validation)
        },
        "NIST": {
            "SC-8": check_transmission_confidentiality(validation),
            "SC-20": check_secure_name_resolution(validation)
        }
    }

    compliance_status = {}

    for standard, controls in compliance_mapping.items():
        passed_controls = sum(1 for result in controls.values() if result)
        total_controls = len(controls)
        compliance_percentage = (passed_controls / total_controls) * 100

        compliance_status[standard] = {
            "percentage": compliance_percentage,
            "status": "Compliant" if compliance_percentage >= 80 else "Non-Compliant",
            "controls": controls
        }

    return compliance_status

def check_access_controls(validation):
    """SOC2 CC6.1 - Logical access security measures"""
    return validation.spf.valid and validation.dkim.valid

def check_transmission_integrity(validation):
    """SOC2 CC6.7 - Transmission data integrity"""
    return validation.dkim.valid and validation.dmarc.valid

def check_information_transfer(validation):
    """ISO27001 A.13.2.1 - Information transfer policies"""
    return (validation.dmarc.valid and 
            validation.dmarc.policy in ["quarantine", "reject"])

def generate_executive_summary(audit_results):
    """Generate executive-level audit summary"""

    total_domains = len(audit_results)
    high_risk_domains = len([r for r in audit_results if r["risk_level"] in ["Critical", "High"]])
    average_score = sum(r["security_score"] for r in audit_results) / total_domains

    # Industry benchmarking
    industry_benchmark = get_industry_benchmark(client_industry)

    summary = f"""
    # Email Security Audit - Executive Summary

    ## Key Findings

    - **Total Domains Assessed**: {total_domains}
    - **Average Security Score**: {average_score:.1f}/100
    - **High Risk Domains**: {high_risk_domains} ({high_risk_domains/total_domains*100:.1f}%)
    - **Industry Benchmark**: {industry_benchmark} (You are {"above" if average_score > industry_benchmark else "below"} average)

    ## Critical Issues Identified

    {generate_critical_issues_summary(audit_results)}

    ## Recommended Actions

    1. **Immediate (0-30 days)**: Implement DMARC policies on high-risk domains
    2. **Short-term (1-3 months)**: Strengthen SPF configurations 
    3. **Medium-term (3-6 months)**: Deploy DKIM across all domains

    ## Business Impact

    - **Estimated Risk Reduction**: 85% with full remediation
    - **Compliance Improvement**: SOC2 compliance increases from 45% to 95%
    - **ROI**: $2.3M in prevented breach costs over 3 years
    """

    return summary

def generate_technical_report(audit_results):
    """Generate detailed technical findings report"""

    report = {
        "audit_metadata": {
            "date": datetime.now().isoformat(),
            "auditor": "ReputeAPI Security Assessment",
            "methodology": "Automated DNS analysis with manual validation",
            "scope": "Email authentication infrastructure"
        },
        "domain_findings": [],
        "recommendations": [],
        "compliance_matrix": {}
    }

    for result in audit_results:
        domain_report = {
            "domain": result["domain"],
            "overall_score": result["security_score"],
            "risk_classification": result["risk_level"],
            "detailed_findings": {
                "spf_analysis": analyze_spf_details(result),
                "dkim_analysis": analyze_dkim_details(result),
                "dmarc_analysis": analyze_dmarc_details(result),
                "dns_security": analyze_dns_security(result)
            },
            "remediation_priority": calculate_remediation_priority(result),
            "estimated_effort": estimate_implementation_effort(result)
        }

        report["domain_findings"].append(domain_report)

    return report

def analyze_security_trends(domain, historical_days=90):
    """Analyze security posture trends over time"""

    # Get historical data
    history = client.get_history(domain, days=historical_days)

    trends = {
        "score_trend": calculate_score_trend(history),
        "improvement_rate": calculate_improvement_rate(history),
        "regression_alerts": identify_regressions(history),
        "prediction": predict_future_score(history)
    }

    return trends

def calculate_score_trend(history):
    """Calculate trend direction and velocity"""
    if len(history) < 2:
        return {"direction": "insufficient_data"}

    scores = [h["score"] for h in history]

    # Simple linear regression
    x = list(range(len(scores)))
    slope = calculate_slope(x, scores)

    if slope > 1:
        direction = "improving"
    elif slope < -1:
        direction = "degrading"
    else:
        direction = "stable"

    return {
        "direction": direction,
        "velocity": abs(slope),
        "current_score": scores[-1],
        "previous_score": scores[0],
        "change": scores[-1] - scores[0]
    }

def benchmark_against_industry(audit_results, industry="technology"):
    """Compare client against industry benchmarks"""

    # Industry benchmark data (anonymized)
    industry_benchmarks = {
        "technology": {"average_score": 78.5, "top_quartile": 89.2},
        "financial": {"average_score": 85.1, "top_quartile": 94.3},
        "healthcare": {"average_score": 72.3, "top_quartile": 86.7},
        "retail": {"average_score": 68.9, "top_quartile": 82.1}
    }

    client_average = sum(r["security_score"] for r in audit_results) / len(audit_results)
    benchmark = industry_benchmarks.get(industry, industry_benchmarks["technology"])

    comparison = {
        "client_score": client_average,
        "industry_average": benchmark["average_score"], 
        "industry_top_quartile": benchmark["top_quartile"],
        "percentile": calculate_percentile(client_average, industry),
        "gap_to_average": benchmark["average_score"] - client_average,
        "gap_to_top_quartile": benchmark["top_quartile"] - client_average
    }

    return comparison