"""HTTP API v2 handler — Bearer-authenticated roster PUT/DELETE. PUT /roster upsert name, extension, slack_user_id, optional email DELETE /roster/{extension} delete the row (204 even if it was already gone) Auth is an app-level Bearer token stored in Secrets Manager. Do not log the Authorization header, the token, or the request body. """ from __future__ import annotations import base64 import hmac import json import logging import os import unicodedata from typing import Any import shared.sentry_init # noqa: F401 from shared.schedule import ShiftSchedule from shared.secrets import get_secret logger = logging.getLogger() logger.setLevel(logging.INFO) PUT_FIELDS = ("name", "extension", "slack_user_id") OPTIONAL_PUT_FIELDS = ("email",) MAX_BODY_BYTES = 4096 MAX_NAME_LEN = 128 MAX_EXTENSION_LEN = 16 MAX_SLACK_ID_LEN = 64 MAX_EMAIL_LEN = 254 ALLOWED_EMAIL_DOMAINS = {"seahaven.com", "seahavenind.com"} _cached_token: str | None = None class AuthError(Exception): """Missing or wrong Bearer token.""" class SecretUnavailable(Exception): """Token secret could not be read.""" def _json_response(status: int, body: dict[str, Any] | None = None) -> dict: if status == 204: return {"statusCode": 204, "headers": {}, "body": ""} payload = {} if body is None else body return { "statusCode": status, "headers": {"Content-Type": "application/json"}, "body": json.dumps(payload, separators=(",", ":")), } def _header(event: dict, name: str) -> str: headers = event.get("headers") or {} if not isinstance(headers, dict): return "" target = name.lower() for key, value in headers.items(): if str(key).lower() == target: return "" if value is None else str(value) return "" def _route(event: dict) -> tuple[str, str]: request_context = event.get("requestContext") or {} http = request_context.get("http") or {} method = str(http.get("method") or event.get("httpMethod") or "").upper() path = str(event.get("rawPath") or http.get("path") or event.get("path") or "") route_key = event.get("routeKey") if isinstance(route_key, str) and " " in route_key: rk_method, rk_path = route_key.split(" ", 1) method = method or rk_method.upper() path = path or rk_path return method, path def _raw_body(event: dict) -> bytes | None: body = event.get("body") if body is None: return b"" try: if event.get("isBase64Encoded"): if isinstance(body, bytes): body = body.decode("ascii") return base64.b64decode(body, validate=True) if isinstance(body, bytes): return body return str(body).encode("utf-8") except (ValueError, UnicodeError): return None def _has_disallowed_chars(value: str, *, allow_space: bool) -> bool: for char in value: if char == " " and allow_space: continue if char.isspace() or unicodedata.category(char).startswith("C"): return True return False def _expected_token() -> str: global _cached_token if _cached_token: return _cached_token secret_id = os.environ["ROSTER_API_TOKEN_SECRET"] try: token = get_secret(secret_id) except Exception: logger.exception("roster api token secret read failed") raise SecretUnavailable from None if not isinstance(token, str): raise SecretUnavailable token = token.strip() if not token: raise SecretUnavailable _cached_token = token return token def _authorize(event: dict) -> None: presented = _header(event, "authorization") if not presented: raise AuthError parts = presented.split(None, 1) if len(parts) != 2 or parts[0].lower() != "bearer" or not parts[1].strip(): raise AuthError token = parts[1].strip() expected = _expected_token() try: matched = hmac.compare_digest(token, expected) except (TypeError, ValueError): raise AuthError from None if not matched: raise AuthError def _validate_email(value: str) -> str: if ( len(value) > MAX_EMAIL_LEN or "@" not in value or _has_disallowed_chars(value, allow_space=False) ): raise ValueError("fields") local, _, domain = value.partition("@") if not local or domain.lower() not in ALLOWED_EMAIL_DOMAINS: raise ValueError("fields") return value.lower() def _validate_put(raw: bytes) -> tuple[str, str, str, str | None]: if len(raw) > MAX_BODY_BYTES: raise ValueError("oversized") try: parsed = json.loads(raw.decode("utf-8")) except (UnicodeDecodeError, json.JSONDecodeError): raise ValueError("invalid json") from None if not isinstance(parsed, dict): raise TypeError("invalid json") allowed = set(PUT_FIELDS) | set(OPTIONAL_PUT_FIELDS) if not set(PUT_FIELDS).issubset(parsed) or not set(parsed).issubset(allowed): raise ValueError("fields") values: dict[str, str] = {} for field in PUT_FIELDS: value = parsed[field] if not isinstance(value, str): raise TypeError("fields") trimmed = value.strip() if not trimmed: raise ValueError("fields") values[field] = trimmed name = values["name"] extension = values["extension"] slack_user_id = values["slack_user_id"] email = None if "email" in parsed: raw_email = parsed["email"] if not isinstance(raw_email, str): raise TypeError("fields") trimmed_email = raw_email.strip() if not trimmed_email: raise ValueError("fields") email = _validate_email(trimmed_email) if len(name) > MAX_NAME_LEN or _has_disallowed_chars(name, allow_space=True): raise ValueError("fields") if ( len(extension) > MAX_EXTENSION_LEN or not extension.isdigit() or _has_disallowed_chars(extension, allow_space=False) ): raise ValueError("fields") if ( len(slack_user_id) > MAX_SLACK_ID_LEN or not slack_user_id.isalnum() or _has_disallowed_chars(slack_user_id, allow_space=False) ): raise ValueError("fields") return name, extension, slack_user_id, email def _delete_extension(event: dict) -> str: params = event.get("pathParameters") or {} if not isinstance(params, dict): raise TypeError("extension") raw = params.get("extension") if not isinstance(raw, str): raise TypeError("extension") extension = raw.strip() if ( not extension or len(extension) > MAX_EXTENSION_LEN or not extension.isdigit() or _has_disallowed_chars(extension, allow_space=False) ): raise ValueError("extension") return extension def handler(event, context): try: method, path = _route(event) logger.info("roster api %s %s", method, path) try: _authorize(event) except AuthError: return _json_response(401, {"error": "unauthorized"}) except SecretUnavailable: return _json_response(503, {"error": "service unavailable"}) if method == "PUT" and (path == "/roster" or path.rstrip("/") == "/roster"): raw = _raw_body(event) if raw is None: return _json_response(400, {"error": "invalid request"}) try: name, extension, slack_user_id, email = _validate_put(raw) except (TypeError, ValueError): return _json_response(400, {"error": "invalid request"}) ShiftSchedule().upsert_roster_entry( extension, name, slack_user_id, email=email ) logger.info("roster upserted extension=%s", extension) return _json_response(200, {"ok": True}) if method == "DELETE" and path.startswith("/roster/"): try: extension = _delete_extension(event) except (TypeError, ValueError): return _json_response(400, {"error": "invalid request"}) ShiftSchedule().remove_roster_entry(extension) logger.info("roster deleted extension=%s", extension) return _json_response(204) return _json_response(405, {"error": "method not allowed"}) except Exception: logger.exception("roster api unexpected failure") return _json_response(500, {"error": "internal error"})