microsoft-365-service/app/credentials.py

import base64
import binascii
import json
from datetime import datetime, timezone
from typing import Annotated

from cryptography.exceptions import InvalidTag
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
from cryptography.hazmat.primitives.kdf.hkdf import HKDF
from fastapi import Header, HTTPException, status

from .config import Settings, settings


CREDENTIAL_VERSION = "v1"
_HKDF_SALT = b"microsoft-365-service.credentials.v1"


def _b64url_decode(value: str) -> bytes:
    padding = "=" * (-len(value) % 4)
    try:
        return base64.urlsafe_b64decode((value + padding).encode("ascii"))
    except (binascii.Error, UnicodeEncodeError) as exc:
        raise ValueError("invalid base64url") from exc


def _derive_key(shared_secret: str, header_name: str) -> bytes:
    return HKDF(
        algorithm=hashes.SHA256(),
        length=32,
        salt=_HKDF_SALT,
        info=header_name.encode("utf-8"),
    ).derive(shared_secret.encode("utf-8"))


def decode_credential_header(encoded_value: str, *, header_name: str, shared_secret: str) -> str:
    parts = encoded_value.split(".")
    if len(parts) != 3 or parts[0] != CREDENTIAL_VERSION:
        raise ValueError("unsupported credential encoding version")

    nonce = _b64url_decode(parts[1])
    ciphertext = _b64url_decode(parts[2])
    if len(nonce) != 12:
        raise ValueError("invalid nonce length")

    key = _derive_key(shared_secret, header_name)
    aad = header_name.encode("utf-8")
    plaintext = AESGCM(key).decrypt(nonce, ciphertext, aad)
    payload = json.loads(plaintext.decode("utf-8"))

    expires_at = payload.get("expires_at")
    if expires_at:
        expires_at_datetime = datetime.fromisoformat(expires_at.replace("Z", "+00:00"))
        if expires_at_datetime <= datetime.now(timezone.utc):
            raise ValueError("credential value is expired")

    value = payload.get("value")
    if not isinstance(value, str) or not value:
        raise ValueError("credential payload must contain a non-empty value")
    return value


async def get_request_settings(
    tenant_id: Annotated[
        str | None,
        Header(
            alias="X-MS365-Tenant-Id",
            description="Encoded value of MS365_TENANT_ID, the Microsoft Entra tenant id. Format: v1.<base64url nonce>.<base64url ciphertext+tag>.",
        ),
    ] = None,
    client_id: Annotated[
        str | None,
        Header(
            alias="X-MS365-Client-Id",
            description="Encoded value of MS365_CLIENT_ID, the Microsoft Entra application client id. Format: v1.<base64url nonce>.<base64url ciphertext+tag>.",
        ),
    ] = None,
    client_secret: Annotated[
        str | None,
        Header(
            alias="X-MS365-Client-Secret",
            description="Encoded value of MS365_CLIENT_SECRET, the Microsoft Entra application client secret. Format: v1.<base64url nonce>.<base64url ciphertext+tag>.",
        ),
    ] = None,
    credential_version: Annotated[
        str | None,
        Header(
            alias="X-MS365-Credential-Version",
            description="Credential encoding version. Currently supported value: v1.",
        ),
    ] = None,
) -> Settings:
    header_values = {
        "X-MS365-Tenant-Id": tenant_id,
        "X-MS365-Client-Id": client_id,
        "X-MS365-Client-Secret": client_secret,
    }
    provided = {name: value for name, value in header_values.items() if value}
    if not provided:
        return settings

    missing = [name for name, value in header_values.items() if not value]
    if missing:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail={"message": "Incomplete Microsoft 365 credential headers.", "missing_headers": missing},
        )

    if credential_version != CREDENTIAL_VERSION:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail=f"X-MS365-Credential-Version must be {CREDENTIAL_VERSION}.",
        )

    if not settings.credential_encoding_secret:
        raise HTTPException(
            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
            detail="MS365_CREDENTIAL_ENCODING_SECRET is not configured.",
        )

    try:
        decoded = {
            name: decode_credential_header(
                value,
                header_name=name,
                shared_secret=settings.credential_encoding_secret,
            )
            for name, value in provided.items()
        }
    except (InvalidTag, ValueError, json.JSONDecodeError, UnicodeDecodeError) as exc:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Microsoft 365 credential headers are malformed or cannot be authenticated.",
        ) from exc

    return settings.with_credentials(
        tenant_id=decoded["X-MS365-Tenant-Id"],
        client_id=decoded["X-MS365-Client-Id"],
        client_secret=decoded["X-MS365-Client-Secret"],
    )