Google Workspace Cli

#!/usr/bin/env python3
"""
Auth Setup Guide - Generate Google Workspace API authentication setup documentation.

Creates step-by-step setup guides for OAuth 2.0, service accounts, and
domain-wide delegation for Google Workspace API access.

Author: Claude Skills Engineering Team
License: MIT
"""

import argparse
import json
import sys
from dataclasses import dataclass, asdict
from typing import List, Dict, Optional


API_SCOPES = {
    "admin": {
        "name": "Admin SDK Directory API",
        "scopes": [
            "https://www.googleapis.com/auth/admin.directory.user",
            "https://www.googleapis.com/auth/admin.directory.user.readonly",
            "https://www.googleapis.com/auth/admin.directory.group",
            "https://www.googleapis.com/auth/admin.directory.orgunit",
        ],
        "description": "Manage users, groups, and organizational units.",
    },
    "drive": {
        "name": "Google Drive API",
        "scopes": [
            "https://www.googleapis.com/auth/drive",
            "https://www.googleapis.com/auth/drive.readonly",
            "https://www.googleapis.com/auth/drive.file",
            "https://www.googleapis.com/auth/drive.metadata.readonly",
        ],
        "description": "Access and manage Google Drive files and folders.",
    },
    "gmail": {
        "name": "Gmail API",
        "scopes": [
            "https://www.googleapis.com/auth/gmail.readonly",
            "https://www.googleapis.com/auth/gmail.send",
            "https://www.googleapis.com/auth/gmail.settings.basic",
            "https://www.googleapis.com/auth/gmail.labels",
        ],
        "description": "Read, send, and manage Gmail messages and settings.",
    },
    "calendar": {
        "name": "Google Calendar API",
        "scopes": [
            "https://www.googleapis.com/auth/calendar",
            "https://www.googleapis.com/auth/calendar.readonly",
            "https://www.googleapis.com/auth/calendar.events",
        ],
        "description": "Access and manage Google Calendar events.",
    },
    "sheets": {
        "name": "Google Sheets API",
        "scopes": [
            "https://www.googleapis.com/auth/spreadsheets",
            "https://www.googleapis.com/auth/spreadsheets.readonly",
        ],
        "description": "Read and write Google Sheets data.",
    },
    "reports": {
        "name": "Admin Reports API",
        "scopes": [
            "https://www.googleapis.com/auth/admin.reports.audit.readonly",
            "https://www.googleapis.com/auth/admin.reports.usage.readonly",
        ],
        "description": "Access audit logs and usage reports.",
    },
}


@dataclass
class SetupStep:
    """A setup step."""
    number: int
    title: str
    instructions: List[str]
    notes: List[str]


def generate_oauth_guide(scope_keys: List[str], project_name: str) -> List[SetupStep]:
    """Generate OAuth 2.0 setup guide."""
    selected_scopes = []
    for key in scope_keys:
        if key in API_SCOPES:
            selected_scopes.extend(API_SCOPES[key]["scopes"])

    steps = [
        SetupStep(
            number=1,
            title="Create Google Cloud Project",
            instructions=[
                "Go to https://console.cloud.google.com/",
                "Click 'Select a project' > 'New Project'",
                f"Enter project name: '{project_name}'",
                "Click 'Create'",
                "Wait for project creation to complete",
            ],
            notes=["Use a descriptive name that identifies this integration."],
        ),
        SetupStep(
            number=2,
            title="Enable Required APIs",
            instructions=[
                "Go to APIs & Services > Library",
            ] + [f"Search for and enable: {API_SCOPES[k]['name']}" for k in scope_keys if k in API_SCOPES],
            notes=["Each API must be explicitly enabled before use."],
        ),
        SetupStep(
            number=3,
            title="Configure OAuth Consent Screen",
            instructions=[
                "Go to APIs & Services > OAuth consent screen",
                "Select 'Internal' (for Workspace users only) or 'External'",
                f"App name: '{project_name}'",
                "Add support email and developer contact",
                "Click 'Save and Continue'",
                "Add scopes:",
            ] + [f"  - {scope}" for scope in selected_scopes],
            notes=[
                "Internal type limits access to your organization only (recommended).",
                "External type requires verification for production use.",
            ],
        ),
        SetupStep(
            number=4,
            title="Create OAuth Client ID",
            instructions=[
                "Go to APIs & Services > Credentials",
                "Click '+ Create Credentials' > 'OAuth client ID'",
                "Application type: 'Desktop app' (for CLI) or 'Web application'",
                f"Name: '{project_name} OAuth Client'",
                "Click 'Create'",
                "Download the JSON credentials file",
                "Save as 'credentials.json' in your project directory",
            ],
            notes=[
                "IMPORTANT: Never commit credentials.json to version control.",
                "Add credentials.json to .gitignore immediately.",
            ],
        ),
        SetupStep(
            number=5,
            title="Test Authentication",
            instructions=[
                "Install the Google client library:",
                "  pip install google-auth google-auth-oauthlib google-api-python-client",
                "Run your first API call to trigger the OAuth flow",
                "Complete the browser-based authorization",
                "Token will be saved locally for future use",
            ],
            notes=["First run will open a browser for authorization."],
        ),
    ]

    return steps


def generate_service_account_guide(scope_keys: List[str], project_name: str,
                                     domain_delegation: bool = False) -> List[SetupStep]:
    """Generate service account setup guide."""
    selected_scopes = []
    for key in scope_keys:
        if key in API_SCOPES:
            selected_scopes.extend(API_SCOPES[key]["scopes"])

    steps = [
        SetupStep(
            number=1,
            title="Create Google Cloud Project",
            instructions=[
                "Go to https://console.cloud.google.com/",
                "Click 'Select a project' > 'New Project'",
                f"Enter project name: '{project_name}'",
                "Click 'Create'",
            ],
            notes=[],
        ),
        SetupStep(
            number=2,
            title="Enable Required APIs",
            instructions=[
                "Go to APIs & Services > Library",
            ] + [f"Enable: {API_SCOPES[k]['name']}" for k in scope_keys if k in API_SCOPES],
            notes=[],
        ),
        SetupStep(
            number=3,
            title="Create Service Account",
            instructions=[
                "Go to IAM & Admin > Service Accounts",
                "Click '+ Create Service Account'",
                f"Name: '{project_name}-service-account'",
                "Click 'Create and Continue'",
                "Grant role: 'Project > Editor' (or more restrictive)",
                "Click 'Continue' then 'Done'",
            ],
            notes=["Use the most restrictive role that satisfies your requirements."],
        ),
        SetupStep(
            number=4,
            title="Create Service Account Key",
            instructions=[
                "Click on the service account you just created",
                "Go to 'Keys' tab",
                "Click 'Add Key' > 'Create new key'",
                "Select JSON format",
                "Click 'Create'",
                "Save the downloaded JSON key file securely",
            ],
            notes=[
                "CRITICAL: This key grants full service account access. Store securely.",
                "Never commit key files to version control.",
                "Rotate keys regularly (every 90 days recommended).",
            ],
        ),
    ]

    if domain_delegation:
        steps.append(SetupStep(
            number=5,
            title="Configure Domain-Wide Delegation",
            instructions=[
                "Copy the service account's Client ID (numeric)",
                "Go to Google Workspace Admin Console > Security > API Controls",
                "Click 'Manage Domain Wide Delegation'",
                "Click 'Add New'",
                "Enter the Client ID",
                "Add the following OAuth scopes:",
            ] + [f"  {scope}" for scope in selected_scopes] + [
                "Click 'Authorize'",
            ],
            notes=[
                "Domain-wide delegation allows the service account to impersonate any user.",
                "Only grant the minimum scopes required.",
                "This requires super admin access to the Admin Console.",
            ],
        ))

        steps.append(SetupStep(
            number=6,
            title="Test Domain-Wide Delegation",
            instructions=[
                "Install dependencies:",
                "  pip install google-auth google-api-python-client",
                "Create a test script that impersonates a user:",
                "  from google.oauth2 import service_account",
                "  credentials = service_account.Credentials.from_service_account_file(",
                "      'key.json',",
                f"      scopes={selected_scopes[:2]},",
                "      subject='admin@yourdomain.com'",
                "  )",
                "Run the test script to verify delegation works",
            ],
            notes=["Replace 'admin@yourdomain.com' with a real user in your domain."],
        ))
    else:
        steps.append(SetupStep(
            number=5,
            title="Test Service Account",
            instructions=[
                "Install dependencies:",
                "  pip install google-auth google-api-python-client",
                "Test authentication with the key file",
                "Verify API calls succeed",
            ],
            notes=[],
        ))

    return steps


def format_text_guide(steps: List[SetupStep], method: str, scope_keys: List[str]) -> str:
    """Format guide as text."""
    lines = []
    lines.append("=" * 60)
    lines.append(f"GOOGLE WORKSPACE API SETUP GUIDE")
    lines.append(f"Method: {method.upper()}")
    lines.append("=" * 60)

    lines.append("\nRequired APIs and Scopes:")
    for key in scope_keys:
        if key in API_SCOPES:
            api = API_SCOPES[key]
            lines.append(f"\n  {api['name']}")
            lines.append(f"  {api['description']}")
            for scope in api["scopes"]:
                lines.append(f"    - {scope}")

    lines.append("\n" + "-" * 60)
    lines.append("SETUP STEPS")
    lines.append("-" * 60)

    for step in steps:
        lines.append(f"\n## Step {step.number}: {step.title}")
        for inst in step.instructions:
            lines.append(f"  {inst}")
        if step.notes:
            lines.append("")
            for note in step.notes:
                lines.append(f"  NOTE: {note}")

    lines.append("\n" + "=" * 60)
    return "\n".join(lines)


def format_json_guide(steps: List[SetupStep], method: str, scope_keys: List[str]) -> str:
    """Format guide as JSON."""
    apis = {}
    for key in scope_keys:
        if key in API_SCOPES:
            apis[key] = API_SCOPES[key]

    return json.dumps({
        "method": method,
        "apis": apis,
        "steps": [asdict(s) for s in steps],
        "total_steps": len(steps),
    }, indent=2)


def main():
    parser = argparse.ArgumentParser(
        description="Generate Google Workspace API authentication setup documentation."
    )
    parser.add_argument("--method", choices=["oauth", "service-account"],
                       required=True, help="Authentication method")
    parser.add_argument("--scopes", required=True,
                       help="Comma-separated API scope keys: admin,drive,gmail,calendar,sheets,reports")
    parser.add_argument("--project", default="my-gws-integration",
                       help="Google Cloud project name")
    parser.add_argument("--domain-delegation", action="store_true",
                       help="Include domain-wide delegation setup (service-account only)")
    parser.add_argument("--format", choices=["text", "json"], default="text", help="Output format")
    parser.add_argument("--list-scopes", action="store_true", help="List available API scopes")
    args = parser.parse_args()

    if args.list_scopes:
        for key, api in API_SCOPES.items():
            print(f"\n{key}: {api['name']}")
            print(f"  {api['description']}")
            for scope in api["scopes"]:
                print(f"    {scope}")
        return

    scope_keys = [s.strip() for s in args.scopes.split(",")]
    invalid = [s for s in scope_keys if s not in API_SCOPES]
    if invalid:
        print(f"Error: Unknown scope keys: {', '.join(invalid)}", file=sys.stderr)
        print(f"Valid keys: {', '.join(API_SCOPES.keys())}", file=sys.stderr)
        sys.exit(2)

    if args.method == "oauth":
        steps = generate_oauth_guide(scope_keys, args.project)
    else:
        steps = generate_service_account_guide(scope_keys, args.project, args.domain_delegation)

    if args.format == "json":
        print(format_json_guide(steps, args.method, scope_keys))
    else:
        print(format_text_guide(steps, args.method, scope_keys))


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
GWS Doctor - Diagnostic tool for common Google Workspace issues.

Checks DNS record format, email configuration, and common integration
patterns. Validates SPF, DKIM, DMARC records and provides fix guidance.

Author: Claude Skills Engineering Team
License: MIT
"""

import argparse
import json
import re
import sys
from dataclasses import dataclass, asdict
from typing import List, Dict, Optional, Any


@dataclass
class DiagnosticResult:
    """A diagnostic check result."""
    check: str
    status: str  # pass, fail, warn, skip
    category: str
    message: str
    detail: Optional[str]
    fix: Optional[str]


GOOGLE_MX_RECORDS = [
    "ASPMX.L.GOOGLE.COM",
    "ALT1.ASPMX.L.GOOGLE.COM",
    "ALT2.ASPMX.L.GOOGLE.COM",
    "ALT3.ASPMX.L.GOOGLE.COM",
    "ALT4.ASPMX.L.GOOGLE.COM",
]

SPF_PATTERN = re.compile(r'v=spf1\s+.*include:_spf\.google\.com\s+.*(?:~all|-all)')
DMARC_PATTERN = re.compile(r'v=DMARC1;\s*p=(\w+)')
DKIM_PATTERN = re.compile(r'v=DKIM1;\s*k=rsa;\s*p=\S+')


class GWSDiagnostics:
    """Runs diagnostic checks on Google Workspace configuration."""

    def __init__(self, config: Dict[str, Any], checks: List[str]):
        self.config = config
        self.checks = checks
        self.results: List[DiagnosticResult] = []

    def run(self) -> List[DiagnosticResult]:
        """Run selected diagnostics."""
        check_map = {
            "dns": self._check_dns,
            "email": self._check_email,
            "security": self._check_security,
            "integration": self._check_integration,
            "consistency": self._check_consistency,
        }

        for check_name in self.checks:
            if check_name == "all":
                for func in check_map.values():
                    func()
                break
            elif check_name in check_map:
                check_map[check_name]()

        return self.results

    def _check_dns(self):
        """Check DNS record configurations."""
        dns = self.config.get("dns", {})

        # MX Records
        mx_records = dns.get("mx_records", [])
        if not mx_records:
            self.results.append(DiagnosticResult(
                check="MX Records",
                status="fail",
                category="dns",
                message="No MX records configured.",
                detail="MX records are required for email delivery.",
                fix="Add Google Workspace MX records: ASPMX.L.GOOGLE.COM (priority 1), "
                    "ALT1.ASPMX.L.GOOGLE.COM (priority 5), etc.",
            ))
        else:
            google_mx_found = any(
                any(gmx in str(mx).upper() for gmx in GOOGLE_MX_RECORDS)
                for mx in mx_records
            )
            if google_mx_found:
                self.results.append(DiagnosticResult(
                    check="MX Records",
                    status="pass",
                    category="dns",
                    message="Google Workspace MX records found.",
                    detail=f"MX records: {', '.join(str(mx) for mx in mx_records[:3])}",
                    fix=None,
                ))
            else:
                self.results.append(DiagnosticResult(
                    check="MX Records",
                    status="warn",
                    category="dns",
                    message="MX records present but Google Workspace records not detected.",
                    detail=f"Current MX: {', '.join(str(mx) for mx in mx_records[:3])}",
                    fix="Verify MX records point to Google: ASPMX.L.GOOGLE.COM",
                ))

        # SPF Record
        spf = dns.get("spf_record", "")
        if not spf:
            self.results.append(DiagnosticResult(
                check="SPF Record",
                status="fail",
                category="dns",
                message="No SPF record configured.",
                detail="SPF validates email sender identity.",
                fix='Add TXT record: "v=spf1 include:_spf.google.com ~all"',
            ))
        elif SPF_PATTERN.search(spf):
            mechanism = "~all (softfail)" if "~all" in spf else "-all (hardfail)"
            self.results.append(DiagnosticResult(
                check="SPF Record",
                status="pass",
                category="dns",
                message=f"SPF record correctly includes Google and uses {mechanism}.",
                detail=spf,
                fix=None,
            ))
        elif "_spf.google.com" in spf:
            self.results.append(DiagnosticResult(
                check="SPF Record",
                status="warn",
                category="dns",
                message="SPF record includes Google but may have formatting issues.",
                detail=spf,
                fix='Ensure format is: "v=spf1 include:_spf.google.com ~all"',
            ))
        else:
            self.results.append(DiagnosticResult(
                check="SPF Record",
                status="fail",
                category="dns",
                message="SPF record does not include Google Workspace.",
                detail=spf,
                fix='Add "include:_spf.google.com" to your SPF record.',
            ))

        # DKIM
        dkim = dns.get("dkim_record", "")
        if not dkim:
            self.results.append(DiagnosticResult(
                check="DKIM Record",
                status="fail",
                category="dns",
                message="No DKIM record configured.",
                detail="DKIM signs outgoing emails to prevent tampering.",
                fix="Enable DKIM in Admin Console > Apps > Google Workspace > Gmail > Authenticate email. "
                    "Then add the CNAME or TXT record to your DNS.",
            ))
        elif DKIM_PATTERN.search(dkim) or "CNAME" in str(dkim):
            self.results.append(DiagnosticResult(
                check="DKIM Record",
                status="pass",
                category="dns",
                message="DKIM record found.",
                detail=f"DKIM: {str(dkim)[:80]}...",
                fix=None,
            ))

        # DMARC
        dmarc = dns.get("dmarc_record", "")
        if not dmarc:
            self.results.append(DiagnosticResult(
                check="DMARC Record",
                status="fail",
                category="dns",
                message="No DMARC record configured.",
                detail="DMARC enforces SPF and DKIM policies.",
                fix='Add TXT record at _dmarc.yourdomain.com: '
                    '"v=DMARC1; p=quarantine; rua=mailto:dmarc-reports@yourdomain.com"',
            ))
        else:
            dmarc_match = DMARC_PATTERN.search(dmarc)
            if dmarc_match:
                policy = dmarc_match.group(1).lower()
                if policy == "none":
                    self.results.append(DiagnosticResult(
                        check="DMARC Record",
                        status="warn",
                        category="dns",
                        message="DMARC policy is 'none' (monitoring only, no enforcement).",
                        detail=dmarc,
                        fix="After monitoring period, change p=none to p=quarantine or p=reject.",
                    ))
                else:
                    self.results.append(DiagnosticResult(
                        check="DMARC Record",
                        status="pass",
                        category="dns",
                        message=f"DMARC record with '{policy}' policy found.",
                        detail=dmarc,
                        fix=None,
                    ))

    def _check_email(self):
        """Check email configuration."""
        email = self.config.get("email", {})

        # Routing
        routing = email.get("routing", "")
        if routing:
            self.results.append(DiagnosticResult(
                check="Email Routing",
                status="pass" if routing == "direct" else "warn",
                category="email",
                message=f"Email routing: {routing}.",
                detail="Direct routing is simplest. Third-party routing adds complexity.",
                fix=None if routing == "direct" else "Review routing configuration for necessity.",
            ))

        # Spam filtering
        spam = email.get("spam_filtering", True)
        if not spam:
            self.results.append(DiagnosticResult(
                check="Spam Filtering",
                status="fail",
                category="email",
                message="Spam filtering appears to be disabled.",
                detail=None,
                fix="Enable spam filtering in Gmail > Spam, phishing, and malware.",
            ))
        else:
            self.results.append(DiagnosticResult(
                check="Spam Filtering",
                status="pass",
                category="email",
                message="Spam filtering is enabled.",
                detail=None,
                fix=None,
            ))

    def _check_security(self):
        """Check security settings."""
        security = self.config.get("security", {})

        # 2FA
        tfa = security.get("2fa_enforcement", False)
        self.results.append(DiagnosticResult(
            check="2FA Enforcement",
            status="pass" if tfa else "fail",
            category="security",
            message=f"2-Step Verification enforcement: {'enabled' if tfa else 'disabled'}.",
            detail=None,
            fix=None if tfa else "Enable 2FA enforcement in Admin Console > Security > Authentication.",
        ))

        # Session control
        session = security.get("session_duration_hours", 24)
        if session and int(session) > 12:
            self.results.append(DiagnosticResult(
                check="Session Duration",
                status="warn",
                category="security",
                message=f"Session duration ({session}h) is longer than recommended (12h).",
                detail=None,
                fix="Reduce session duration to 12 hours or less.",
            ))
        else:
            self.results.append(DiagnosticResult(
                check="Session Duration",
                status="pass",
                category="security",
                message=f"Session duration ({session}h) is within recommended range.",
                detail=None,
                fix=None,
            ))

    def _check_integration(self):
        """Check common integration patterns."""
        integrations = self.config.get("integrations", {})

        # SSO
        sso = integrations.get("sso_enabled", False)
        sso_provider = integrations.get("sso_provider", "")
        if sso:
            self.results.append(DiagnosticResult(
                check="SSO Configuration",
                status="pass",
                category="integration",
                message=f"SSO enabled via {sso_provider or 'configured provider'}.",
                detail=None,
                fix=None,
            ))
        else:
            self.results.append(DiagnosticResult(
                check="SSO Configuration",
                status="warn",
                category="integration",
                message="SSO is not configured.",
                detail="SSO provides centralized authentication control.",
                fix="Consider configuring SSO if using an identity provider (Okta, Azure AD, etc.).",
            ))

        # LDAP/Directory Sync
        dir_sync = integrations.get("directory_sync", False)
        if dir_sync:
            self.results.append(DiagnosticResult(
                check="Directory Sync",
                status="pass",
                category="integration",
                message="Directory synchronization is active.",
                detail=None,
                fix=None,
            ))

    def _check_consistency(self):
        """Check for configuration consistency issues."""
        security = self.config.get("security", {})
        drive = self.config.get("drive", {})

        # Inconsistency: strict auth but open sharing
        tfa = security.get("2fa_enforcement", False)
        sharing = drive.get("external_sharing", "")
        if tfa and sharing in ("allowed", "unrestricted"):
            self.results.append(DiagnosticResult(
                check="Auth-Sharing Consistency",
                status="warn",
                category="consistency",
                message="2FA enforced but external sharing is unrestricted.",
                detail="Strong auth with open sharing may still leak data.",
                fix="Consider restricting external sharing to match the strict auth posture.",
            ))

        # Inconsistency: advanced mobile but no 2FA
        mobile = self.config.get("devices", {}).get("mobile_management", "")
        if mobile == "advanced" and not tfa:
            self.results.append(DiagnosticResult(
                check="Mobile-Auth Consistency",
                status="warn",
                category="consistency",
                message="Advanced mobile management without 2FA enforcement.",
                detail="Mobile management is less effective without mandatory 2FA.",
                fix="Enable 2FA enforcement to complement mobile management.",
            ))


def generate_sample_config() -> Dict[str, Any]:
    """Generate sample configuration for testing."""
    return {
        "dns": {
            "mx_records": ["ASPMX.L.GOOGLE.COM", "ALT1.ASPMX.L.GOOGLE.COM"],
            "spf_record": "v=spf1 include:_spf.google.com ~all",
            "dkim_record": "",
            "dmarc_record": "v=DMARC1; p=none; rua=mailto:dmarc@example.com",
        },
        "email": {
            "routing": "direct",
            "spam_filtering": True,
        },
        "security": {
            "2fa_enforcement": False,
            "session_duration_hours": 24,
        },
        "drive": {
            "external_sharing": "allowed",
        },
        "devices": {
            "mobile_management": "basic",
        },
        "integrations": {
            "sso_enabled": False,
            "directory_sync": False,
        },
    }


def format_text(results: List[DiagnosticResult]) -> str:
    """Format as human-readable text."""
    lines = []
    lines.append("=" * 60)
    lines.append("GOOGLE WORKSPACE DIAGNOSTIC REPORT")
    lines.append("=" * 60)

    passed = sum(1 for r in results if r.status == "pass")
    failed = sum(1 for r in results if r.status == "fail")
    warned = sum(1 for r in results if r.status == "warn")

    lines.append(f"\nChecks: {len(results)} total")
    lines.append(f"  PASS: {passed}  |  FAIL: {failed}  |  WARN: {warned}")
    lines.append("-" * 60)

    for status_label, status_key in [("FAILURES", "fail"), ("WARNINGS", "warn"), ("PASSED", "pass")]:
        group = [r for r in results if r.status == status_key]
        if not group:
            continue

        icon = {"fail": "FAIL", "warn": "WARN", "pass": "PASS"}[status_key]
        lines.append(f"\n[{status_label}]")
        for r in group:
            lines.append(f"  [{icon}] {r.check} ({r.category})")
            lines.append(f"    {r.message}")
            if r.detail:
                lines.append(f"    Detail: {r.detail}")
            if r.fix:
                lines.append(f"    Fix: {r.fix}")
            lines.append("")

    health = "HEALTHY" if failed == 0 else "NEEDS ATTENTION" if failed <= 2 else "CRITICAL"
    lines.append(f"Overall Health: {health}")
    lines.append("=" * 60)
    return "\n".join(lines)


def format_json(results: List[DiagnosticResult]) -> str:
    """Format as JSON."""
    return json.dumps({
        "results": [asdict(r) for r in results],
        "summary": {
            "total": len(results),
            "pass": sum(1 for r in results if r.status == "pass"),
            "fail": sum(1 for r in results if r.status == "fail"),
            "warn": sum(1 for r in results if r.status == "warn"),
        }
    }, indent=2)


def main():
    parser = argparse.ArgumentParser(
        description="Diagnostic tool for common Google Workspace configuration issues."
    )
    parser.add_argument("--config", "-c", help="Path to GWS config JSON")
    parser.add_argument("--sample", action="store_true", help="Run against sample config")
    parser.add_argument("--check", default="all",
                       help="Checks to run: all,dns,email,security,integration,consistency")
    parser.add_argument("--format", choices=["text", "json"], default="text", help="Output format")
    parser.add_argument("--generate-sample", action="store_true", help="Output sample config")
    args = parser.parse_args()

    if args.generate_sample:
        print(json.dumps(generate_sample_config(), indent=2))
        return

    if args.sample:
        config = generate_sample_config()
    elif args.config:
        from pathlib import Path
        path = Path(args.config)
        if not path.exists():
            print(f"Error: File not found: {args.config}", file=sys.stderr)
            sys.exit(2)
        try:
            config = json.loads(path.read_text())
        except json.JSONDecodeError as e:
            print(f"Error: Invalid JSON: {e}", file=sys.stderr)
            sys.exit(2)
    else:
        parser.error("Provide --config or --sample")
        return

    checks = [c.strip() for c in args.check.split(",")]
    diagnostics = GWSDiagnostics(config, checks)
    results = diagnostics.run()

    if args.format == "json":
        print(format_json(results))
    else:
        print(format_text(results))

    if any(r.status == "fail" for r in results):
        sys.exit(1)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Google Workspace Audit - Audit GWS configurations for security best practices.

Analyzes exported Workspace configuration JSON for security gaps including
2FA enforcement, password policies, sharing settings, and admin roles.

Author: Claude Skills Engineering Team
License: MIT
"""

import argparse
import json
import sys
from dataclasses import dataclass, asdict
from pathlib import Path
from typing import List, Dict, Any, Optional


@dataclass
class Finding:
    """An audit finding."""
    severity: str  # critical, high, medium, low
    category: str
    setting: str
    current_value: str
    expected_value: str
    message: str
    recommendation: str


# Default security policy expectations
SECURITY_POLICIES = {
    "2fa_enforcement": {
        "path": "security.2fa_enforcement",
        "expected": True,
        "severity": "critical",
        "category": "authentication",
        "message": "2-Step Verification is not enforced for all users.",
        "recommendation": "Enable 2-Step Verification enforcement in Admin Console > Security > Authentication.",
    },
    "password_min_length": {
        "path": "security.password_policy.min_length",
        "expected_min": 12,
        "severity": "critical",
        "category": "authentication",
        "message": "Password minimum length is below recommended threshold.",
        "recommendation": "Set minimum password length to 12+ characters in Admin Console > Security > Password management.",
    },
    "password_require_numbers": {
        "path": "security.password_policy.require_numbers",
        "expected": True,
        "severity": "high",
        "category": "authentication",
        "message": "Password policy does not require numbers.",
        "recommendation": "Enable number requirement in password policy.",
    },
    "password_require_symbols": {
        "path": "security.password_policy.require_symbols",
        "expected": True,
        "severity": "medium",
        "category": "authentication",
        "message": "Password policy does not require symbols.",
        "recommendation": "Enable symbol requirement in password policy.",
    },
    "external_sharing": {
        "path": "drive.external_sharing",
        "expected": "restricted",
        "severity": "high",
        "category": "data_protection",
        "message": "External file sharing is not restricted.",
        "recommendation": "Set Drive external sharing to 'restricted' or 'off' for sensitive OUs.",
    },
    "link_sharing_default": {
        "path": "drive.link_sharing_default",
        "expected": "restricted",
        "severity": "high",
        "category": "data_protection",
        "message": "Default link sharing allows broad access.",
        "recommendation": "Set default link sharing to 'Restricted' (only people with access).",
    },
    "third_party_apps": {
        "path": "security.third_party_app_access",
        "expected": "restricted",
        "severity": "high",
        "category": "app_security",
        "message": "Third-party app access is not restricted.",
        "recommendation": "Limit third-party app access. Review and whitelist approved apps only.",
    },
    "mobile_management": {
        "path": "devices.mobile_management",
        "expected": "advanced",
        "severity": "medium",
        "category": "device_security",
        "message": "Mobile device management is not set to advanced.",
        "recommendation": "Enable advanced mobile management for device encryption and remote wipe.",
    },
    "admin_recovery": {
        "path": "security.admin_recovery_options",
        "expected": True,
        "severity": "medium",
        "category": "authentication",
        "message": "Admin account recovery options may be misconfigured.",
        "recommendation": "Ensure super admin accounts have recovery phone and email configured.",
    },
    "session_duration": {
        "path": "security.session_duration_hours",
        "expected_max": 12,
        "severity": "medium",
        "category": "authentication",
        "message": "Session duration is longer than recommended.",
        "recommendation": "Set session duration to 12 hours or less for web sessions.",
    },
    "email_whitelist_only": {
        "path": "gmail.restrict_delivery",
        "expected": False,
        "severity": "low",
        "category": "email",
        "message": "Email delivery restrictions check.",
        "recommendation": "Review email delivery settings for appropriate restriction level.",
    },
}


def get_nested_value(data: Dict, path: str, default=None) -> Any:
    """Get a nested dictionary value by dot-separated path."""
    keys = path.split(".")
    current = data
    for key in keys:
        if isinstance(current, dict) and key in current:
            current = current[key]
        else:
            return default
    return current


class WorkspaceAuditor:
    """Audits Google Workspace configuration against security best practices."""

    def __init__(self, config: Dict[str, Any]):
        self.config = config
        self.findings: List[Finding] = []

    def audit(self) -> List[Finding]:
        """Run full security audit."""
        self._audit_security_policies()
        self._audit_admin_roles()
        self._audit_api_access()
        self._audit_email_security()
        return self.findings

    def _audit_security_policies(self):
        """Audit against defined security policies."""
        for policy_name, policy in SECURITY_POLICIES.items():
            path = policy["path"]
            current = get_nested_value(self.config, path)

            if current is None:
                self.findings.append(Finding(
                    severity=policy["severity"],
                    category=policy["category"],
                    setting=path,
                    current_value="not configured",
                    expected_value=str(policy.get("expected", policy.get("expected_min", policy.get("expected_max", "")))),
                    message=policy["message"],
                    recommendation=policy["recommendation"],
                ))
                continue

            # Boolean check
            if "expected" in policy and isinstance(policy["expected"], bool):
                if current != policy["expected"]:
                    self.findings.append(Finding(
                        severity=policy["severity"],
                        category=policy["category"],
                        setting=path,
                        current_value=str(current),
                        expected_value=str(policy["expected"]),
                        message=policy["message"],
                        recommendation=policy["recommendation"],
                    ))

            # Minimum check
            elif "expected_min" in policy:
                try:
                    if int(current) < policy["expected_min"]:
                        self.findings.append(Finding(
                            severity=policy["severity"],
                            category=policy["category"],
                            setting=path,
                            current_value=str(current),
                            expected_value=f">= {policy['expected_min']}",
                            message=policy["message"],
                            recommendation=policy["recommendation"],
                        ))
                except (ValueError, TypeError):
                    pass

            # Maximum check
            elif "expected_max" in policy:
                try:
                    if int(current) > policy["expected_max"]:
                        self.findings.append(Finding(
                            severity=policy["severity"],
                            category=policy["category"],
                            setting=path,
                            current_value=str(current),
                            expected_value=f"<= {policy['expected_max']}",
                            message=policy["message"],
                            recommendation=policy["recommendation"],
                        ))
                except (ValueError, TypeError):
                    pass

            # String check
            elif "expected" in policy and isinstance(policy["expected"], str):
                if str(current).lower() != policy["expected"].lower():
                    self.findings.append(Finding(
                        severity=policy["severity"],
                        category=policy["category"],
                        setting=path,
                        current_value=str(current),
                        expected_value=policy["expected"],
                        message=policy["message"],
                        recommendation=policy["recommendation"],
                    ))

    def _audit_admin_roles(self):
        """Audit admin role assignments."""
        admins = get_nested_value(self.config, "admin.super_admins", [])
        if isinstance(admins, list):
            if len(admins) > 5:
                self.findings.append(Finding(
                    severity="high",
                    category="access_control",
                    setting="admin.super_admins",
                    current_value=f"{len(admins)} super admins",
                    expected_value="<= 5 super admins",
                    message=f"Too many super admin accounts ({len(admins)}). Excessive privileges increase risk.",
                    recommendation="Reduce super admin count. Use delegated admin roles for specific tasks.",
                ))

            if len(admins) < 2:
                self.findings.append(Finding(
                    severity="medium",
                    category="access_control",
                    setting="admin.super_admins",
                    current_value=f"{len(admins)} super admin(s)",
                    expected_value=">= 2 super admins",
                    message="Too few super admins. Single point of failure for admin access.",
                    recommendation="Designate at least 2 super admin accounts for redundancy.",
                ))

    def _audit_api_access(self):
        """Audit API and OAuth app access."""
        api_clients = get_nested_value(self.config, "security.api_clients", [])
        if isinstance(api_clients, list):
            for client in api_clients:
                if isinstance(client, dict):
                    scopes = client.get("scopes", [])
                    name = client.get("name", "unknown")
                    if isinstance(scopes, list) and len(scopes) > 10:
                        self.findings.append(Finding(
                            severity="medium",
                            category="app_security",
                            setting=f"security.api_clients.{name}",
                            current_value=f"{len(scopes)} scopes",
                            expected_value="minimal scopes",
                            message=f"API client '{name}' has {len(scopes)} scopes. Excessive scope grants increase risk.",
                            recommendation=f"Review scopes for '{name}' and remove unnecessary permissions.",
                        ))

    def _audit_email_security(self):
        """Audit email-related security settings."""
        spoofing = get_nested_value(self.config, "gmail.spoofing_protection")
        if spoofing is not None and not spoofing:
            self.findings.append(Finding(
                severity="high",
                category="email",
                setting="gmail.spoofing_protection",
                current_value="disabled",
                expected_value="enabled",
                message="Email spoofing protection is not enabled.",
                recommendation="Enable spoofing protection in Gmail > Safety > Spoofing and authentication.",
            ))

        dmarc = get_nested_value(self.config, "dns.dmarc_policy")
        if dmarc and str(dmarc).lower() == "none":
            self.findings.append(Finding(
                severity="high",
                category="email",
                setting="dns.dmarc_policy",
                current_value="none",
                expected_value="quarantine or reject",
                message="DMARC policy is set to 'none', providing no email protection.",
                recommendation="Set DMARC policy to 'quarantine' or 'reject' after monitoring period.",
            ))


def generate_sample_config() -> Dict[str, Any]:
    """Generate a sample configuration for testing."""
    return {
        "security": {
            "2fa_enforcement": False,
            "password_policy": {
                "min_length": 8,
                "require_numbers": True,
                "require_symbols": False,
            },
            "third_party_app_access": "unrestricted",
            "session_duration_hours": 24,
            "admin_recovery_options": True,
            "api_clients": [],
        },
        "drive": {
            "external_sharing": "allowed",
            "link_sharing_default": "anyone_with_link",
        },
        "devices": {
            "mobile_management": "basic",
        },
        "admin": {
            "super_admins": ["admin@example.com"],
        },
        "gmail": {
            "spoofing_protection": True,
            "restrict_delivery": False,
        },
        "dns": {
            "dmarc_policy": "none",
        },
    }


def format_text(findings: List[Finding]) -> str:
    """Format as human-readable text."""
    lines = []
    lines.append("=" * 60)
    lines.append("GOOGLE WORKSPACE SECURITY AUDIT REPORT")
    lines.append("=" * 60)

    by_severity = {}
    for f in findings:
        by_severity.setdefault(f.severity, []).append(f)

    lines.append(f"\nTotal findings: {len(findings)}")
    for sev in ["critical", "high", "medium", "low"]:
        count = len(by_severity.get(sev, []))
        if count:
            lines.append(f"  {sev.upper()}: {count}")

    lines.append("-" * 60)

    for sev in ["critical", "high", "medium", "low"]:
        group = by_severity.get(sev, [])
        if not group:
            continue
        lines.append(f"\n[{sev.upper()}]")
        for f in group:
            lines.append(f"  [{f.category}] {f.setting}")
            lines.append(f"    Issue: {f.message}")
            lines.append(f"    Current: {f.current_value} | Expected: {f.expected_value}")
            lines.append(f"    Fix: {f.recommendation}")
            lines.append("")

    if not findings:
        lines.append("\nNo security issues found. Configuration follows best practices.")

    lines.append("=" * 60)
    return "\n".join(lines)


def format_json(findings: List[Finding]) -> str:
    """Format as JSON."""
    return json.dumps({
        "findings": [asdict(f) for f in findings],
        "summary": {
            "total": len(findings),
            "critical": sum(1 for f in findings if f.severity == "critical"),
            "high": sum(1 for f in findings if f.severity == "high"),
            "medium": sum(1 for f in findings if f.severity == "medium"),
            "low": sum(1 for f in findings if f.severity == "low"),
        }
    }, indent=2)


def main():
    parser = argparse.ArgumentParser(
        description="Audit Google Workspace configuration for security best practices."
    )
    parser.add_argument("--config", "-c", help="Path to GWS configuration JSON export")
    parser.add_argument("--sample", action="store_true", help="Run audit against sample config for demo")
    parser.add_argument("--format", choices=["text", "json"], default="text", help="Output format")
    parser.add_argument("--generate-sample", action="store_true",
                       help="Output sample config JSON for testing")
    args = parser.parse_args()

    if args.generate_sample:
        print(json.dumps(generate_sample_config(), indent=2))
        return

    if args.sample:
        config = generate_sample_config()
    elif args.config:
        path = Path(args.config)
        if not path.exists():
            print(f"Error: File not found: {args.config}", file=sys.stderr)
            sys.exit(2)
        try:
            config = json.loads(path.read_text())
        except json.JSONDecodeError as e:
            print(f"Error: Invalid JSON: {e}", file=sys.stderr)
            sys.exit(2)
    else:
        parser.error("Provide --config or --sample")
        return

    auditor = WorkspaceAuditor(config)
    findings = auditor.audit()

    if args.format == "json":
        print(format_json(findings))
    else:
        print(format_text(findings))

    if any(f.severity == "critical" for f in findings):
        sys.exit(1)


if __name__ == "__main__":
    main()