Control Monitoring.

"""CCM-01 SSO enforced on vendor apps.

Every active vendor in tier 1 or 2 must sit behind the identity provider with
SSO enforced and local password login turned off. Local logins skip MFA and
survive offboarding, so this is the control that keeps the others honest.
"""

from datetime import date

CONTROL = {
    "id": "CCM-01",
    "name": "SSO enforced",
    "question": "Is every in-scope vendor app behind SSO, with password login off?",
    "why": "A local login skips MFA and is missed when someone leaves. Vendors say they support SSO in their SOC 2; this checks that we actually turned it on.",
    "sources": ["Vendor inventory", "Identity provider apps"],
    "frequency": "Daily",
    "response": "Password login switched back on for an SSO app is reverted automatically. Anything else opens a ticket for the app owner.",
    "frameworks": ["SOC 2 CC6.1", "ISO 27001 A.8.5", "NIST CSF PR.AA-03"],
}

IN_SCOPE_TIERS = (1, 2)


def check(state, today):
    findings = []
    tested = 0
    for vid, v in sorted(state["vendors"].items()):
        if v["status"] != "active" or v["tier"] not in IN_SCOPE_TIERS:
            continue
        if date.fromisoformat(v["contract_end"]) < today:
            continue  # ended vendors belong to CCM-04
        tested += 1
        app = state["idp_apps"].get(v.get("app") or "")
        if app is None or not app["active"]:
            findings.append({
                "key": f"CCM-01:{vid}:no-idp",
                "title": f"{v['name']} is not connected to the identity provider",
                "severity": "High",
                "evidence": {"vendor": v["name"], "tier": v["tier"], "data": v["data"], "idp_app": None},
                "fix": f"Create the SSO app for {v['name']}, move users over, then turn off password login.",
            })
            continue
        if not app["sso_enforced"]:
            findings.append({
                "key": f"CCM-01:{vid}:not-enforced",
                "title": f"{v['name']} allows sign in without SSO",
                "severity": "High",
                "evidence": {"vendor": v["name"], "idp_app": v["app"], "sso_enforced": False, "password_login": app["password_login"]},
                "fix": f"Turn on SSO enforcement for {v['name']} and remove local accounts.",
            })
        elif app["password_login"]:
            findings.append({
                "key": f"CCM-01:{vid}:password-login",
                "title": f"Password login was switched back on for {v['name']}",
                "severity": "Critical",
                "evidence": {"vendor": v["name"], "idp_app": v["app"], "sso_enforced": True, "password_login": True},
                "fix": "Reverted automatically. Ask the admin who changed it whether a break glass account is needed.",
                # Safe to undo on our own: SSO already works for everyone.
                "auto_fix": [{"op": "set", "path": ["idp_apps", v["app"], "password_login"], "value": False}],
            })
    return tested, findings

"""CCM-02 User access drift.

Compares every vendor app's user list with the HR system. Flags accounts that
belong to people who left, contractors past their end date, and admin rights
that were never approved.
"""

from datetime import date

CONTROL = {
    "id": "CCM-02",
    "name": "User access drift",
    "question": "Does anyone still have vendor app access they should not?",
    "why": "Offboarding usually removes SSO, but local accounts and admin rights inside vendor apps drift. Quarterly access reviews find this months later; a daily diff finds it the next morning.",
    "sources": ["HR system", "Vendor app user lists", "Approved admin list"],
    "frequency": "Daily",
    "response": "Opens a ticket for the app owner with the account and the reason. Never removes access on its own, because the owner may need to move data first.",
    "frameworks": ["SOC 2 CC6.2", "SOC 2 CC6.3", "ISO 27001 A.5.18", "NIST CSF PR.AA-05"],
}


def check(state, today):
    findings = []
    tested = 0
    people = state["hris"]
    for app_id, users in sorted(state["app_users"].items()):
        vendor = next((v for v in state["vendors"].values() if v.get("app") == app_id), None)
        if vendor is None or vendor["status"] != "active":
            continue  # ended vendors belong to CCM-04
        approved = set(state["approved_admins"].get(app_id, []))
        for email, acct in sorted(users.items()):
            tested += 1
            person = people.get(email)
            base = {"app": vendor["name"], "account": email, "role": acct["role"]}
            if person is None:
                findings.append({
                    "key": f"CCM-02:{app_id}:{email}:unknown",
                    "title": f"{email} in {vendor['name']} is not in the HR system",
                    "severity": "High",
                    "evidence": {**base, "hr_record": None},
                    "fix": "Find who owns the account. Remove it or link it to a named person.",
                })
            elif person["status"] == "terminated":
                findings.append({
                    "key": f"CCM-02:{app_id}:{email}:left",
                    "title": f"{person['name']} left on {person['end_date']} but still has {vendor['name']}",
                    "severity": "High",
                    "evidence": {**base, "hr_status": "terminated", "end_date": person["end_date"]},
                    "fix": f"Remove the {vendor['name']} account and check its last sign in.",
                })
            elif person["type"] == "contractor" and person.get("end_date") and date.fromisoformat(person["end_date"]) < today:
                findings.append({
                    "key": f"CCM-02:{app_id}:{email}:contract-ended",
                    "title": f"Contractor {person['name']} ended {person['end_date']}, {vendor['name']} access still active",
                    "severity": "High",
                    "evidence": {**base, "type": "contractor", "end_date": person["end_date"]},
                    "fix": "Remove access, or get the contract extension and update the end date in HR.",
                })
            elif acct["role"] == "admin" and email not in approved:
                findings.append({
                    "key": f"CCM-02:{app_id}:{email}:unapproved-admin",
                    "title": f"{person['name']} is an admin in {vendor['name']} without approval",
                    "severity": "Medium",
                    "evidence": {**base, "approved_admins": sorted(approved)},
                    "fix": "Approve it with a reason, or drop the account back to a standard user.",
                })
    return tested, findings

"""CCM-03 Shadow vendors.

Any app a user connects to Google Workspace with access to mail, files or the
directory is a vendor holding company data. If it is not in the vendor
inventory, nobody has assessed it.
"""

CONTROL = {
    "id": "CCM-03",
    "name": "Shadow vendors",
    "question": "Are there apps reading company data that never went through vendor review?",
    "why": "One click on Allow gives a new AI tool read access to someone's Drive. That is a new vendor with no assessment, no contract and no exit plan.",
    "sources": ["Google Workspace OAuth grants", "Vendor inventory"],
    "frequency": "Daily",
    "response": "Opens a vendor intake ticket and messages the users who granted it. Blocking the app is a decision for the risk owner.",
    "frameworks": ["SOC 2 CC9.2", "ISO 27001 A.5.19", "NIST CSF GV.SC-04"],
}

# Scopes that reach company data. Calendar free/busy and sign in only do not count.
SENSITIVE = {
    "gmail.readonly": "read all mail",
    "gmail.modify": "read and change mail",
    "drive": "read and change all Drive files",
    "drive.readonly": "read all Drive files",
    "admin.directory.user.readonly": "read the staff directory",
}


def check(state, today):
    known = {v["oauth_client"] for v in state["vendors"].values() if v.get("oauth_client") and v["status"] == "active"}
    findings = []
    tested = 0
    for client, g in sorted(state["oauth_grants"].items()):
        tested += 1
        risky = [s for s in g["scopes"] if s in SENSITIVE]
        if not risky or client in known:
            continue
        findings.append({
            "key": f"CCM-03:{client}",
            "title": f"{g['app']} can {SENSITIVE[risky[0]]} and is not a reviewed vendor",
            "severity": "High" if len(g["users"]) > 1 or "gmail" in risky[0] else "Medium",
            "evidence": {"app": g["app"], "client_id": client, "scopes": risky, "granted_by": g["users"], "first_grant": g["granted"]},
            "fix": f"Start vendor intake for {g['app']}. If it is rejected, revoke the grant for all users.",
        })
    return tested, findings

"""CCM-04 Vendor offboarding.

When a vendor contract ends, everything that connects us to them should be
gone: the SSO app, user accounts, API keys and OAuth grants.
"""

from datetime import date

CONTROL = {
    "id": "CCM-04",
    "name": "Vendor offboarding",
    "question": "When a vendor contract ends, is every connection to them actually closed?",
    "why": "Contracts end in the procurement system. API keys and SSO apps live somewhere else, so they keep working long after anyone remembers the vendor.",
    "sources": ["Vendor inventory (contract dates)", "Identity provider apps", "Vendor app user lists", "API key register", "Google Workspace OAuth grants"],
    "frequency": "Daily",
    "response": "Opens one ticket per vendor listing everything still connected, and updates it as items close.",
    "frameworks": ["SOC 2 CC9.2", "ISO 27001 A.5.22", "NIST CSF GV.SC-10"],
}


def check(state, today):
    findings = []
    tested = 0
    for vid, v in sorted(state["vendors"].items()):
        ended = v["status"] == "offboarded" or date.fromisoformat(v["contract_end"]) < today
        if not ended:
            continue
        tested += 1
        open_items = []
        app = state["idp_apps"].get(v.get("app") or "")
        if app and app["active"]:
            open_items.append(f"SSO app {v['app']} still active")
        users = state["app_users"].get(v.get("app") or "", {})
        if users:
            open_items.append(f"{len(users)} user accounts")
        keys = [k for k, key in state["api_keys"].items() if key["vendor"] == vid and key["active"]]
        open_items += [f"API key {k} active" for k in keys]
        if v.get("oauth_client") in state["oauth_grants"]:
            open_items.append("OAuth grant in Google Workspace")
        if open_items:
            findings.append({
                "key": f"CCM-04:{vid}",
                "title": f"{v['name']} contract ended {v['contract_end']} but is still connected",
                "severity": "High" if keys or v["data"] == "Confidential" else "Medium",
                "evidence": {"vendor": v["name"], "contract_end": v["contract_end"], "still_open": open_items},
                "fix": "Revoke the keys first, then deactivate the SSO app and remove accounts. Ask the vendor for data deletion confirmation.",
            })
    return tested, findings