railiance-platform/scripts/mas_evidence_iam_grant.py

#!/usr/bin/env python3
"""Compare and correct the MAS publisher IAM rule.

A temporary Scaleway Console session is accepted on stdin only with --execute,
in the same shape the Coulomb identity bootstrap uses. Preview contacts nothing.
Receipts record policy names, project names, and permission sets. Session
tokens, API access keys, and secret keys are never written.
"""

from __future__ import annotations

import importlib.util
import json
import os
import pwd
import sys
from pathlib import Path
from urllib.parse import urlencode

ROOT = Path("/home/worsch/railiance-platform")
COULOMB_APPLICATION = "681d00bf-b965-4323-b657-97bfde423511"
MAS_APPLICATION = "25a4cbca-220e-40c0-9af5-6c7713000174"
COULOMB_POLICY_NAME = "coulomb-social-backup-primary"
POLICY_NAME = "mas-evidence-backup-primary"
PERMISSIONS = (
    "ObjectStorageBucketsRead",
    "ObjectStorageObjectsRead",
    "ObjectStorageObjectsWrite",
    "ObjectStorageObjectsDelete",
)
_FORBIDDEN_RECEIPT_FIELDS = frozenset({
    "access_key",
    "secret_key",
    "session_token",
    "token",
    "secret",
})


class Refused(RuntimeError):
    pass


def _recovery():
    spec = importlib.util.spec_from_file_location(
        "application_recovery_identities",
        ROOT / "scripts/application_recovery_identities.py",
    )
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


def receipt_path():
    home = Path(pwd.getpwuid(os.getuid()).pw_dir)
    return home / ".local/state/railiance/receipts/mas-evidence-iam-grant.json"


def _write_receipt(value):
    path = receipt_path()
    path.parent.mkdir(mode=0o700, exist_ok=True)
    path.write_text(json.dumps(value, indent=2, sort_keys=True) + "\n", encoding="utf-8")
    path.chmod(0o600)


def assert_public(value):
    if isinstance(value, dict):
        if _FORBIDDEN_RECEIPT_FIELDS & set(value):
            raise Refused("receipt_field_refused")
        for item in value.values():
            assert_public(item)
    elif isinstance(value, list):
        for item in value:
            assert_public(item)


def inventory(api, path, key, **filters):
    rows = []
    for page in range(1, 101):
        value = api("GET", path + "?" + urlencode({**filters, "page_size": 100, "page": page}))
        batch = value[key]
        if not isinstance(batch, list):
            raise Refused("invalid_provider_inventory")
        rows.extend(batch)
        count = int(value["total_count"])
        if len(rows) == count:
            return rows
        if not batch or len(rows) > count:
            break
    raise Refused("complete_inventory_required")


def _rule_view(rule, names):
    project_ids = [item for item in (rule.get("project_ids") or []) if isinstance(item, str)]
    return {
        "permission_set_names": sorted(
            item for item in (rule.get("permission_set_names") or []) if isinstance(item, str)
        ),
        "project_ids": project_ids,
        "project_names": [names.get(item, "") for item in project_ids],
        "organization_scope": bool(rule.get("organization_id")),
        "deny": bool(rule.get("deny")),
        "condition_present": bool(rule.get("condition")),
    }


def _publisher_shape(rule):
    return (
        set(rule.get("permission_set_names") or []) == set(PERMISSIONS)
        and not rule.get("organization_id")
        and not rule.get("condition")
        and not rule.get("deny")
    )


def _matches(rule, project_id):
    return _publisher_shape(rule) and list(rule.get("project_ids") or []) == [project_id]


def _bucket_project(policies, names):
    """The live bucket grant is the Coulomb policy of that name, not the recovery project."""
    chosen = [item for item in policies if item.get("name") == COULOMB_POLICY_NAME]
    if len(chosen) != 1 or len(chosen[0]["_rules"]) != 1:
        raise Refused("coulomb_backup_project_unresolved")
    rule = chosen[0]["_rules"][0]
    project_ids = [item for item in (rule.get("project_ids") or []) if isinstance(item, str)]
    if not _publisher_shape(rule) or len(project_ids) != 1 or project_ids[0] not in names:
        raise Refused("coulomb_backup_project_unresolved")
    return project_ids[0], names[project_ids[0]]


def _policies_for(api, application_id, organization_id, names):
    found = []
    for policy in inventory(
        api, "/iam/v1alpha1/policies", "policies", organization_id=organization_id,
        application_id=application_id,
    ):
        if policy.get("application_id") != application_id:
            continue
        rules = inventory(api, "/iam/v1alpha1/rules", "rules", policy_id=policy["id"])
        found.append({
            "id": policy.get("id"),
            "name": policy.get("name"),
            "editable": bool(policy.get("editable", True)),
            "rules": [_rule_view(rule, names) for rule in rules],
            "_rules": rules,
        })
    return found


def _key_view(keys, project_id):
    projects = sorted({
        item.get("default_project_id") for item in keys
        if isinstance(item.get("default_project_id"), str)
    })
    return {
        "count": len(keys),
        "default_project_ids": projects,
        "preferred_project_matches": len(keys) == 1 and keys[0].get("default_project_id") == project_id,
    }


def _desired_rule(project_id):
    return {"permission_set_names": list(PERMISSIONS), "project_ids": [project_id]}


def align(api):
    """Read both applications. Correct one direct MAS rule when that is unambiguous."""
    coulomb = api("GET", "/iam/v1alpha1/applications/" + COULOMB_APPLICATION)
    mas = api("GET", "/iam/v1alpha1/applications/" + MAS_APPLICATION)
    if coulomb.get("name") != "coulomb-social-backup-primary":
        raise Refused("coulomb_application_binding_refused")
    if mas.get("name") != POLICY_NAME:
        raise Refused("mas_application_binding_refused")
    organization = coulomb.get("organization_id")
    if not organization or organization != mas.get("organization_id"):
        raise Refused("organization_binding_refused")
    available = {
        item.get("name") for item in inventory(
            api, "/iam/v1alpha1/permission-sets", "permission_sets", organization_id=organization,
        )
    }
    if any(name not in available for name in PERMISSIONS):
        raise Refused("required_permission_sets_unavailable")
    projects = inventory(api, "/account/v3/projects", "projects", organization_id=organization)
    names = {item.get("id"): item.get("name") for item in projects if isinstance(item.get("id"), str)}
    coulomb_policies = _policies_for(api, COULOMB_APPLICATION, organization, names)
    project_id, project_name = _bucket_project(coulomb_policies, names)
    mas_policies = _policies_for(api, MAS_APPLICATION, organization, names)
    groups = [
        item.get("name") for item in inventory(
            api, "/iam/v1alpha1/groups", "groups", organization_id=organization,
        )
        if MAS_APPLICATION in (item.get("application_ids") or [])
    ]
    keys = [
        item for item in inventory(
            api, "/iam/v1alpha1/api-keys", "api_keys", organization_id=organization,
            application_id=MAS_APPLICATION,
        )
        if item.get("application_id") == MAS_APPLICATION
    ]
    changed = False
    preferred_updated = False
    if len(mas_policies) > 1:
        outcome = "multiple_policies_require_review"
    elif len(mas_policies) == 1 and len(mas_policies[0]["_rules"]) > 1:
        outcome = "multiple_rules_require_review"
    elif len(mas_policies) == 1 and not mas_policies[0]["editable"]:
        outcome = "policy_not_editable"
    elif len(mas_policies) == 1 and all(_matches(rule, project_id) for rule in mas_policies[0]["_rules"]):
        outcome = "already_matches"
    elif len(mas_policies) == 1:
        policy_id = mas_policies[0]["id"]
        api("PUT", "/iam/v1alpha1/rules", {"policy_id": policy_id, "rules": [_desired_rule(project_id)]})
        changed = True
        outcome = "rule_replaced"
    else:
        api("POST", "/iam/v1alpha1/policies", {
            "name": POLICY_NAME,
            "description": "MAS evidence publisher; one project; bucket policy withholds object delete",
            "organization_id": organization,
            "application_id": MAS_APPLICATION,
            "rules": [_desired_rule(project_id)],
        })
        changed = True
        outcome = "policy_created"
    if outcome in {"already_matches", "rule_replaced", "policy_created"} and len(keys) == 1:
        if keys[0].get("default_project_id") != project_id:
            access_key = keys[0].get("access_key")
            if not isinstance(access_key, str) or not access_key:
                raise Refused("api_key_binding_refused")
            api("PATCH", "/iam/v1alpha1/api-keys/" + access_key, {"default_project_id": project_id})
            preferred_updated = True
    mas_policies = _policies_for(api, MAS_APPLICATION, organization, names)
    keys = [
        item for item in inventory(
            api, "/iam/v1alpha1/api-keys", "api_keys", organization_id=organization,
            application_id=MAS_APPLICATION,
        )
        if item.get("application_id") == MAS_APPLICATION
    ]
    if outcome in {"rule_replaced", "policy_created"}:
        if len(mas_policies) != 1 or not all(_matches(rule, project_id) for rule in mas_policies[0]["_rules"]):
            raise Refused("effective_rule_refused")
    receipt = {
        "status": "aligned" if outcome in {"already_matches", "rule_replaced", "policy_created"} else "review",
        "outcome": outcome,
        "changed": changed,
        "project_name": project_name,
        "project_source": COULOMB_POLICY_NAME,
        "project_id": project_id,
        "permission_sets": list(PERMISSIONS),
        "coulomb_policies": [_public_policy(item) for item in coulomb_policies],
        "mas_policies": [_public_policy(item) for item in mas_policies],
        "mas_application_groups": groups,
        "mas_keys": _key_view(keys, project_id),
        "preferred_project_updated": preferred_updated,
        "credential_values_printed": False,
    }
    assert_public(receipt)
    return receipt


def _public_policy(policy):
    return {"id": policy["id"], "name": policy["name"], "editable": policy["editable"], "rules": policy["rules"]}


def preview():
    return {
        "status": "preview",
        "coulomb_application": COULOMB_APPLICATION,
        "mas_application": MAS_APPLICATION,
        "project_source": COULOMB_POLICY_NAME,
        "permission_sets": list(PERMISSIONS),
        "provider_contacted": False,
        "credential_values_printed": False,
    }


def main(argv=None):
    args = list(sys.argv[1:] if argv is None else argv)
    if args not in ([], ["--execute"]):
        raise Refused("usage_refused")
    if not args:
        print(json.dumps(preview(), sort_keys=True))
        return 0
    raw = sys.stdin.read(65537)
    if len(raw) > 65536:
        raise Refused("bounded_temporary_auth_required")
    recovery = _recovery()
    try:
        auth = json.loads(raw)
    except json.JSONDecodeError:
        raise Refused("temporary_stdin_auth_required") from None
    try:
        receipt = align(recovery.Transport(auth))
    except recovery.Refused as error:
        raise Refused(str(error)) from None
    except recovery.ProviderError as error:
        print(json.dumps({
            "status": "failed", "http_status": error.status, "credential_values_printed": False,
        }, sort_keys=True))
        return 1
    _write_receipt(receipt)
    print(json.dumps(receipt, sort_keys=True))
    return 0


if __name__ == "__main__":
    try:
        raise SystemExit(main())
    except Refused as error:
        print(json.dumps({
            "status": "failed", "refusal": str(error), "credential_values_printed": False,
        }, sort_keys=True))
        raise SystemExit(1) from None
    except (OSError, KeyError, ValueError, TypeError):
        print(json.dumps({
            "status": "failed", "refusal": "contained_grant_failure", "credential_values_printed": False,
        }, sort_keys=True))
        raise SystemExit(1) from None