認証・権限・統制── MSAL の認証、最小権限、秘密情報、個人情報、監査証跡

"""Sign in with the device code flow (MSAL for Python), reusing a cached account first."""
import os
import sys

import msal

SCOPES = ["User.Read"]


def get_token(app, scopes):
    accounts = app.get_accounts()
    if accounts:
        result = app.acquire_token_silent(scopes, account=accounts[0])
        if result and "access_token" in result:
            return result
    flow = app.initiate_device_flow(scopes=scopes)
    if "user_code" not in flow:
        raise RuntimeError("could not start device flow: %s" % flow.get("error_description"))
    print(flow["message"], file=sys.stderr)  # where to go and which code to enter
    return app.acquire_token_by_device_flow(flow)  # blocks until sign-in, expiry or cancel


def main():
    app = msal.PublicClientApplication(
        os.environ["CLIENT_ID"],
        authority="https://login.microsoftonline.com/" + os.environ["TENANT_ID"],
    )
    result = get_token(app, SCOPES)
    if "access_token" not in result:
        print("sign-in failed:", result.get("error"), result.get("error_description"))
        return 1
    print("signed in; token expires in", result.get("expires_in"), "seconds")  # never print the token
    return 0


if __name__ == "__main__":
    sys.exit(main())

"""Keep the MSAL token cache in an encrypted file outside the repository."""
import os
import sys

import msal
from msal_extensions import PersistedTokenCache, build_encrypted_persistence

SCOPES = ["User.Read"]
# Point this outside the repo (e.g. your user profile) and list the file name in .gitignore
CACHE_PATH = os.environ.get("TOKEN_CACHE_PATH", "token_cache.bin")


def build_app():
    persistence = build_encrypted_persistence(CACHE_PATH)  # raises if encryption is unavailable
    print("cache:", type(persistence).__name__, "encrypted:", persistence.is_encrypted)
    return msal.PublicClientApplication(
        os.environ["CLIENT_ID"],
        authority="https://login.microsoftonline.com/" + os.environ["TENANT_ID"],
        token_cache=PersistedTokenCache(persistence),  # file lock and reload are handled
    )


def main():
    app = build_app()
    accounts = app.get_accounts()
    result = app.acquire_token_silent(SCOPES, account=accounts[0]) if accounts else None
    if not result:
        flow = app.initiate_device_flow(scopes=SCOPES)
        if "user_code" not in flow:
            print("could not start sign-in:", flow.get("error_description"))
            return 1
        print(flow["message"], file=sys.stderr)
        result = app.acquire_token_by_device_flow(flow)
    print("token ready" if "access_token" in result else "failed: %s" % result.get("error"))
    return 0


if __name__ == "__main__":
    sys.exit(main())

"""App-only token with a certificate (no client secret), then read license counts."""
import os
import sys

import msal
import requests

GRAPH = "https://graph.microsoft.com/v1.0"
SCOPES = ["https://graph.microsoft.com/.default"]  # app-only tokens always use /.default


def build_app():
    credential = {"private_key_pfx_path": os.environ["CERT_PATH"]}
    if os.environ.get("CERT_PASSPHRASE"):
        credential["passphrase"] = os.environ["CERT_PASSPHRASE"]
    return msal.ConfidentialClientApplication(
        os.environ["CLIENT_ID"],
        authority="https://login.microsoftonline.com/" + os.environ["TENANT_ID"],
        client_credential=credential,
    )


def seat_rows(payload):
    rows = []
    for sku in payload.get("value", []):
        enabled = (sku.get("prepaidUnits") or {}).get("enabled", 0)
        rows.append((sku.get("skuPartNumber"), sku.get("consumedUnits", 0), enabled))
    return rows


def main():
    result = build_app().acquire_token_for_client(scopes=SCOPES)
    if "access_token" not in result:
        print("token error:", result.get("error"), result.get("error_description"))
        return 1
    resp = requests.get(
        GRAPH + "/subscribedSkus",
        params={"$select": "skuPartNumber,consumedUnits,prepaidUnits"},
        headers={"Authorization": "Bearer " + result["access_token"]},
        timeout=30,
    )
    resp.raise_for_status()
    for name, used, enabled in seat_rows(resp.json()):
        print(f"{name}: {used} of {enabled} assigned")
    return 0


if __name__ == "__main__":
    sys.exit(main())

"""Read license counts from code running in Azure, using its managed identity."""
import asyncio
import os

from azure.identity.aio import ManagedIdentityCredential
from kiota_abstractions.base_request_configuration import RequestConfiguration
from msgraph import GraphServiceClient
from msgraph.generated.subscribed_skus.subscribed_skus_request_builder import (
    SubscribedSkusRequestBuilder,
)

SCOPES = ["https://graph.microsoft.com/.default"]


def seat_summary(skus):
    rows = []
    for s in skus:
        enabled = s.prepaid_units.enabled if s.prepaid_units else 0
        rows.append((s.sku_part_number, s.consumed_units or 0, enabled or 0))
    return rows


async def main():
    # user-assigned identity: set AZURE_CLIENT_ID; system-assigned: leave it unset
    async with ManagedIdentityCredential(client_id=os.environ.get("AZURE_CLIENT_ID")) as cred:
        client = GraphServiceClient(credentials=cred, scopes=SCOPES)
        query = SubscribedSkusRequestBuilder.SubscribedSkusRequestBuilderGetQueryParameters(
            select=["skuPartNumber", "consumedUnits", "prepaidUnits"])
        page = await client.subscribed_skus.get(
            request_configuration=RequestConfiguration(query_parameters=query))
        for name, used, enabled in seat_summary(page.value or []):
            print(f"{name}: {used} of {enabled} assigned")


if __name__ == "__main__":
    asyncio.run(main())

"""Read an API key from Azure Key Vault at run time; never print or log the value."""
import os
import sys

from azure.core.exceptions import ClientAuthenticationError, ResourceNotFoundError
from azure.identity import DefaultAzureCredential
from azure.keyvault.secrets import SecretClient

VAULT_URL = os.environ["KEY_VAULT_URL"]
SECRET_NAME = os.environ.get("SECRET_NAME", "partner-api-key")
KV_SCOPE = "https://vault.azure.net/.default"


def load_secret(credential, name, make_client=SecretClient):
    credential.get_token(KV_SCOPE)  # fail fast on sign-in problems, before calling the vault
    secret = make_client(vault_url=VAULT_URL, credential=credential).get_secret(name)
    return secret.value, secret.properties.version


def main():
    credential = DefaultAzureCredential()  # developer sign-in locally, managed identity in Azure
    try:
        value, version = load_secret(credential, SECRET_NAME)
    except ClientAuthenticationError as e:
        print("sign-in failed:", type(e).__name__)
        return 1
    except ResourceNotFoundError:
        print("secret not found:", SECRET_NAME)
        return 1
    print("loaded", SECRET_NAME, "version", version)  # the value itself is never shown
    # use `value` here, e.g. as a request header for the external service
    return 0


if __name__ == "__main__":
    sys.exit(main())

"""Flag over-broad Microsoft Graph permissions in a per-script permission ledger."""
import json
import os
import re

LEDGER = [
    {"script": "mail_tally.py", "app": "reports", "operation": "read", "type": "delegated",
     "scopes": ["Mail.Read"]},
    {"script": "license_report.py", "app": "reports", "operation": "read", "type": "application",
     "scopes": ["LicenseAssignment.Read.All"]},
    {"script": "list_audit.py", "app": "audit", "operation": "read", "type": "delegated",
     "scopes": ["Sites.ReadWrite.All"]},
    {"script": "dir_dump.py", "app": "audit", "operation": "read", "type": "delegated",
     "scopes": ["Directory.Read.All"]},
]
WRITE = re.compile(r"ReadWrite|\.Send$|\.Write|Manage|FullControl")
BROAD = {"Directory.AccessAsUser.All", "Directory.ReadWrite.All", "Directory.Read.All"}


def review(rows):
    findings = []
    for r in rows:
        for s in r["scopes"]:
            if r["operation"] == "read" and WRITE.search(s):
                findings.append((r["script"], s, "write permission on a read-only script"))
            if s in BROAD:
                findings.append((r["script"], s, "broad directory permission; look for a narrower one"))
    kinds = {}
    for r in rows:
        kinds.setdefault(r["app"], set()).add(r["type"])
    for app, k in sorted(kinds.items()):
        if len(k) > 1:
            findings.append((app, "+".join(sorted(k)), "one app mixes delegated and application permissions"))
    return findings


def main():
    path = os.environ.get("PERM_LEDGER")  # optional JSON file with the same shape as LEDGER
    rows = LEDGER
    if path:
        with open(path, encoding="utf-8") as f:
            rows = json.load(f)
    found = review(rows)
    for f in found:
        print(" | ".join(f))
    print("findings:", len(found))


if __name__ == "__main__":
    main()

"""Retry on 429/503, honoring Retry-After; checked here with fake responses (no network)."""
import time

RETRY_STATUS = {429, 503}


def call_with_retry(send, max_attempts=5, base_delay=1.0, sleep=time.sleep):
    """send() returns an object with .status_code and .headers, e.g. a requests.Response."""
    for attempt in range(1, max_attempts + 1):
        resp = send()
        if resp.status_code not in RETRY_STATUS or attempt == max_attempts:
            return resp
        header = resp.headers.get("Retry-After", "")
        wait = float(header) if header.isdigit() else base_delay * 2 ** (attempt - 1)
        print(f"attempt {attempt}: HTTP {resp.status_code}, waiting {wait:g}s")
        sleep(wait)
    return resp


class FakeResponse:
    def __init__(self, status_code, headers=None):
        self.status_code, self.headers = status_code, headers or {}


def demo():
    queue = [FakeResponse(429, {"Retry-After": "10"}), FakeResponse(503), FakeResponse(200)]
    waits = []
    resp = call_with_retry(lambda: queue.pop(0), sleep=waits.append)
    assert resp.status_code == 200 and waits == [10.0, 2.0], waits

    stuck = [FakeResponse(429) for _ in range(3)]
    waits2 = []
    resp2 = call_with_retry(lambda: stuck.pop(0), max_attempts=3, sleep=waits2.append)
    assert resp2.status_code == 429 and waits2 == [1.0, 2.0], waits2
    print("retried:", len(waits), "then 200;", len(waits2), "then gave up")
    # real use: call_with_retry(lambda: session.get(url, headers=headers, timeout=30))


if __name__ == "__main__":
    demo()

"""JSON Lines audit log for a script run, masking e-mail addresses and bearer tokens."""
import json
import logging
import os
import re
import uuid
from collections import Counter
from datetime import datetime, timezone

EMAIL = re.compile(r"[\w.+-]+@[\w-]+(?:\.[\w-]+)+")
BEARER = re.compile(r"Bearer\s+[\w.~+/=-]+")
SELECT = ["from", "receivedDateTime"]  # ask Graph only for what the tally needs


class RedactFilter(logging.Filter):
    def filter(self, record):
        msg = EMAIL.sub("[email]", record.getMessage())
        record.msg, record.args = BEARER.sub("Bearer [redacted]", msg), None
        return True


def audit_logger(path):
    handler = logging.FileHandler(path, encoding="utf-8")
    handler.addFilter(RedactFilter())
    log = logging.getLogger("audit")
    log.setLevel(logging.INFO)
    log.addHandler(handler)
    return log


def event(log, run_id, action, **fields):
    rec = {"ts": datetime.now(timezone.utc).isoformat(timespec="seconds"), "run_id": run_id,
           "run_by": os.environ.get("RUN_BY", "unknown"), "action": action, **fields}
    log.info(json.dumps(rec, ensure_ascii=False))


def main():
    run_id = str(uuid.uuid4())
    log = audit_logger("out_audit.jsonl")
    with open("samples/messages.json", encoding="utf-8") as f:
        rows = json.load(f)["value"]  # stands in for GET /me/messages?$select=from,receivedDateTime
    event(log, run_id, "graph.get", resource="/me/messages", select=",".join(SELECT),
          client_request_id=str(uuid.uuid4()), count=len(rows))
    domains = Counter(r["from"]["emailAddress"]["address"].split("@")[-1] for r in rows)
    event(log, run_id, "tally", sender_domains=len(domains), messages=sum(domains.values()))
    # demo of a careless line: the filter masks the address and the token before they reach the file
    sender = rows[0]["from"]["emailAddress"]["address"]
    event(log, run_id, "debug", note="sender %s auth Bearer abc.def.ghi" % sender)
    logging.shutdown()
    with open("out_audit.jsonl", encoding="utf-8") as f:
        text = f.read()
    assert "@" not in text and "abc.def" not in text
    print("audit lines:", len(text.splitlines()), "| redacted: ok")
    print(text.splitlines()[-1])


if __name__ == "__main__":
    main()