Niveaux 6-7-8-9 — MDM, Groupes, Logs & Recon

9. Niveau 6 — Appareils Mobiles et Points de Terminaison

9.1 Inventaire MDM

Google Workspace intègre une gestion basique des appareils mobiles (MDM). L'audit MDM révèle souvent des appareils fantômes ou des appareils BYOD non gérés.

def audit_mobile_devices(service):
    """Inventaire complet des appareils mobiles enregistrés."""
    devices = []
    page_token = None
    while True:
        result = service.mobiledevices().list(
            customerId="my_customer",
            pageToken=page_token,
            maxResults=100,
            projection="FULL",
        ).execute()
        devices.extend(result.get("mobiledevices", []))
        page_token = result.get("nextPageToken")
        if not page_token:
            break

    findings = {
        "total": len(devices),
        "by_status": {},
        "old_devices": [],      # Pas de sync depuis > 6 mois
        "risky_devices": [],    # BYOD sur comptes admins
        "unencrypted": [],
    }
    for device in devices:
        last_sync = device.get("lastSync", "")
        owner = device.get("email", "")
        status = device.get("status", "")
        # ... analyse
    return findings

• Appareils non synchronisés depuis > 6 mois : fantômes à supprimer
• Appareils BYOD appartenant à des Super Admins : risque élevé
• Appareils sans chiffrement activé
• Absence de PIN obligatoire
• Absence de remote wipe configuré

9.2 Configuration MDM recommandée

10. Niveau 7 — Groupes et Politiques de Partage

10.1 Audit des groupes

def audit_groups(service, groups_settings_service):
    """Analyse la sécurité des groupes Google."""
    findings = {
        "total_groups": 0,
        "external_members": [],
        "orphan_groups": [],    # Sans owner
        "public_groups": [],    # Accessibles par tous
    }

    page_token = None
    while True:
        result = service.groups().list(
            domain=DOMAIN,
            maxResults=200,
            pageToken=page_token,
        ).execute()
        groups = result.get("groups", [])
        findings["total_groups"] += len(groups)

        for group in groups:
            group_email = group.get("email", "")
            members = service.members().list(
                groupKey=group_email
            ).execute().get("members", [])

            # Membres externes
            ext = [m for m in members if not m.get("email", "").endswith(f"@{DOMAIN}")]
            if ext:
                findings["external_members"].append({
                    "group": group_email,
                    "external": ext,
                })

            # Groupes orphelins (sans owner)
            owners = [m for m in members if m.get("role") == "OWNER"]
            if not owners:
                findings["orphan_groups"].append(group_email)

        page_token = result.get("nextPageToken")
        if not page_token:
            break
    return findings

10.2 Politiques de partage global

11. Niveau 8 — Journalisation, Alertes et Surveillance

11.1 Alert Center

L'Alert Center Google Workspace centralise les alertes de sécurité du tenant.

def audit_alert_center(service, days=90):
    """Récupère et analyse les alertes du Alert Center."""
    from datetime import datetime, timezone, timedelta
    import json

    start = (datetime.now(timezone.utc) - timedelta(days=days)).isoformat()
    alerts = []
    page_token = None

    while True:
        result = service.alerts().list(
            filter=f'createTime > "{start}"',
            pageToken=page_token,
        ).execute()
        alerts.extend(result.get("alerts", []))
        page_token = result.get("nextPageToken")
        if not page_token:
            break

    HIGH_SEVERITY_TYPES = {
        "Suspicious login": "CRITIQUE",
        "Phishing email reported": "ÉLEVÉ",
        "Malware detected": "CRITIQUE",
        "Super admin password reset": "ÉLEVÉ",
        "User suspended (spam)": "ÉLEVÉ",
        "Government-backed attack": "CRITIQUE",
        "Data exfiltration": "CRITIQUE",
    }

    categorized = {}
    for alert in alerts:
        alert_type = alert.get("type", "Unknown")
        severity = HIGH_SEVERITY_TYPES.get(alert_type, "INFO")
        categorized.setdefault(alert_type, []).append({
            "id": alert.get("alertId"),
            "time": alert.get("createTime"),
            "severity": severity,
        })
    return {"total": len(alerts), "by_type": categorized}

11.2 Logs de connexion

def audit_login_logs(service, days=30):
    """Analyse les logs de connexion pour détecter des comportements suspects."""
    from datetime import datetime, timezone, timedelta

    start = (datetime.now(timezone.utc) - timedelta(days=days)).isoformat() + "Z"
    findings = {
        "failed_logins": {},
        "suspicious_logins": [],
        "login_by_country": {},
        "anomalous_ips": [],
    }

    page_token = None
    while True:
        result = service.activities().list(
            userKey="all",
            applicationName="login",
            startTime=start,
            maxResults=1000,
            pageToken=page_token,
        ).execute()

        for activity in result.get("items", []):
            actor = activity.get("actor", {}).get("email", "unknown")
            ip = activity.get("ipAddress", "")

            for event in activity.get("events", []):
                name = event.get("name", "")
                if name == "login_failure":
                    findings["failed_logins"].setdefault(actor, 0)
                    findings["failed_logins"][actor] += 1
                elif name == "suspicious_login":
                    findings["suspicious_logins"].append({
                        "user": actor,
                        "ip": ip,
                        "time": activity.get("id", {}).get("time"),
                    })

        page_token = result.get("nextPageToken")
        if not page_token:
            break
    return findings

• Connexions depuis des pays inhabituels (géolocalisation des IPs)
• Connexions à des heures anormales (2h-5h du matin pour une organisation française)
• Pic d'échecs de connexion sur un compte (> 5 en 24h = brute force potentiel)
• Connexions simultanées depuis des localisations géographiquement impossibles
• Super Admins avec des échecs de connexion répétés

11.3 Logs d'administration

def audit_admin_logs(service, days=30):
    """Analyse les actions d'administration des 30 derniers jours."""
    SENSITIVE_ACTIONS = [
        "CHANGE_USER_PASSWORD",
        "CHANGE_RECOVERY_EMAIL",
        "CHANGE_RECOVERY_PHONE",
        "GRANT_ADMIN_PRIVILEGE",
        "REVOKE_ADMIN_PRIVILEGE",
        "DELETE_USER",
        "SUSPEND_USER",
        "CREATE_SERVICE_ACCOUNT",
        "CHANGE_APPLICATION_SETTING",
        "TOGGLE_SERVICE_ENABLED",
        "DOMAIN_WIDE_DELEGATION_CHANGE",
    ]
    # Implémenter via Reports API applicationName="admin"
    pass

11.4 Alertes recommandées à configurer

12. Niveau 9 — Reconnaissance Passive et Threat Intelligence

12.1 Certificate Transparency (crt.sh)

Les Certificate Transparency Logs enregistrent tous les certificats TLS émis. Ils permettent de découvrir des sous-domaines non documentés.

import requests

def check_certificate_transparency(domain):
    """Découverte de sous-domaines via Certificate Transparency Logs."""
    try:
        resp = requests.get(
            f"https://crt.sh/?q=%.{domain}&output=json",
            timeout=15,
            headers={"Accept": "application/json"},
        )
        certs = resp.json()
        subdomains = set()
        wildcards = []

        for cert in certs:
            for name in cert.get("name_value", "").split("\n"):
                name = name.strip().lower()
                if name.endswith(f".{domain}") or name == domain:
                    if name.startswith("*."):
                        wildcards.append(name)
                    else:
                        subdomains.add(name)
        return {
            "subdomains": sorted(subdomains),
            "wildcard_certs": wildcards,
            "total": len(subdomains),
        }
    except Exception as e:
        return {"error": str(e)}

12.2 Google Dorks — Recherche de données exposées

Les Google Dorks permettent de chercher des données spécifiques indexées par Google. À exécuter manuellement dans un navigateur (les requêtes automatiques violent les ToS Google).

# Documents confidentiels indexés
site:votredomaine.com filetype:pdf "confidentiel"
site:votredomaine.com filetype:xlsx OR filetype:csv

# Documents Google Drive publics
site:drive.google.com "votredomaine.com"
site:docs.google.com "votredomaine.com"

# Leaks de credentials
"votredomaine.com" "api_key" OR "secret_key" OR "password"
intext:"@votredomaine.com" site:pastebin.com

# Code source exposé
site:github.com "votredomaine.com"
site:gitlab.com "votredomaine.com"

# Dumps et backups
"votredomaine.com" filetype:sql OR filetype:bak OR filetype:dump

12.3 Shodan — Exposure Internet

Shodan indexe les services Internet exposés. Sans clé API, l'API gratuite internetdb.shodan.io donne des informations de base.

def check_shodan_free(ip):
    """Interroge l'API Shodan InternetDB (gratuite, sans clé)."""
    try:
        resp = requests.get(
            f"https://internetdb.shodan.io/{ip}",
            timeout=10,
        )
        if resp.status_code == 200:
            data = resp.json()
            return {
                "ports": data.get("ports", []),
                "hostnames": data.get("hostnames", []),
                "cpes": data.get("cpes", []),  # Technologies détectées
                "vulns": data.get("vulns", []),
                "tags": data.get("tags", []),  # ex: "eol-product"
            }
    except Exception as e:
        return {"error": str(e)}

• eol-product : logiciel en fin de vie (PHP 7.x, Apache 2.2, etc.)
• self-signed : certificat auto-signé
• starttls : chiffrement opportuniste (non forcé)
• vpn : service VPN exposé

12.4 HIBP — Have I Been Pwned

HIBP permet de vérifier si des emails d'une organisation apparaissent dans des bases de données de credentials compromis.

def check_hibp_domain(domain, api_key):
    """Vérifie via l'API HIBP Domain Search tous les emails d'un domaine."""
    headers = {
        "hibp-api-key": api_key,
        "User-Agent": "Security-Audit-Tool",
    }
    try:
        resp = requests.get(
            f"https://haveibeenpwned.com/api/v3/breacheddomain/{domain}",
            headers=headers,
            timeout=15,
        )
        if resp.status_code == 200:
            return resp.json()  # {email: [breach1, breach2, ...]}
        elif resp.status_code == 404:
            return {}  # Aucun email compromis
        elif resp.status_code == 402:
            return {"error": "Clé API requise — https://haveibeenpwned.com/API/Key"}
    except Exception as e:
        return {"error": str(e)}

Automatisation de la vérification HIBP

Pour les organisations avec une clé API HIBP, la vérification peut être automatisée sur l'ensemble du domaine :

import requests, time

def hibp_domain_search(domain, api_key):
    """
    HIBP Domain Search — retourne tous les emails compromis du domaine.
    Nécessite un abonnement HIBP (clé API).
    """
    headers = {
        "hibp-api-key": api_key,
        "User-Agent": "WorkspaceSecurityAudit/1.0",
    }
    resp = requests.get(
        f"https://haveibeenpwned.com/api/v3/breacheddomain/{domain}",
        headers=headers,
        timeout=30,
    )
    if resp.status_code == 200:
        # Retourne {email: [breach_name1, breach_name2, ...]}
        data = resp.json()
        # Classifier par sévérité
        stealer_sources = {"Naz.API", "ALIEN TXTBASE Stealer Logs",
                          "Exploit.in", "Operation Endgame", "Cit0day"}
        critical = {
            email: breaches for email, breaches in data.items()
            if any(s in breaches for s in stealer_sources)
        }
        return {"all_compromised": data, "critical_stealer": critical}
    return {"error": f"HTTP {resp.status_code}"}

def hibp_check_recovery_emails(recovery_emails, api_key):
    """Vérifie les emails de récupération individuellement."""
    results = []
    for entry in recovery_emails:
        resp = requests.get(
            f"https://haveibeenpwned.com/api/v3/breachedaccount/{entry['recovery_email']}",
            headers={"hibp-api-key": api_key, "User-Agent": "Audit/1.0"},
            timeout=10,
        )
        if resp.status_code == 200:
            breaches = resp.json()
            results.append({
                "cyrias_account": entry["email"],
                "recovery_email": entry["recovery_email"],
                "breach_count": len(breaches),
                "breaches": [b["Name"] for b in breaches],
                "severity": "CRITIQUE" if len(breaches) > 5 else "MOYEN",
            })
        time.sleep(1.5)  # Rate limiting HIBP : 1 requête/1.5s
    return results

1. Leur rôle dans l'organisation (Super Admin = risque maximal)
2. La présence d'un email de récupération externe
3. Le type de breach (stealer log vs. scraping public)