credentials do hlavicky + crypto + docu

2026-05-28 13:10:01 +02:00
parent f3553dec26
commit 41fb87cefa
5 changed files with 338 additions and 31 deletions
@@ -0,0 +1,141 @@
+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"],
+    )