#!/usr/bin/env python3 """python_api_input_validation.py A practical starter script for secure API input validation. What this script demonstrates: - Type hints as documentation and design guidance - Runtime validation as the actual enforcement layer - A clear separation between raw payload parsing and trusted application data - Safe handling of required fields, optional fields, ranges, and nested structures This script is vendor-neutral and intentionally avoids any framework-specific code. It can be adapted for Flask, FastAPI, Django, Starlette, or custom services. """ from __future__ import annotations import argparse import json import sys from dataclasses import dataclass from datetime import datetime, timezone from decimal import Decimal from typing import Any, Optional from uuid import UUID class ValidationError(ValueError): """Raised when input does not match the expected contract.""" @dataclass(frozen=True) class UserProfileUpdate: """Validated input model for a profile update request.""" user_id: UUID display_name: str age: Optional[int] tags: list[str] timezone_name: Optional[str] credit_limit: Optional[Decimal] requested_at: datetime def _require_dict(value: Any, field_name: str = "payload") -> dict[str, Any]: if not isinstance(value, dict): raise ValidationError(f"{field_name} must be a JSON object") return value def _require_string(value: Any, field_name: str, *, min_len: int = 1, max_len: int = 200) -> str: if not isinstance(value, str): raise ValidationError(f"{field_name} must be a string") text = value.strip() if len(text) < min_len: raise ValidationError(f"{field_name} must not be empty") if len(text) > max_len: raise ValidationError(f"{field_name} must be at most {max_len} characters") return text def _require_optional_string(value: Any, field_name: str, *, max_len: int = 100) -> Optional[str]: if value is None: return None return _require_string(value, field_name, min_len=1, max_len=max_len) def _require_int(value: Any, field_name: str, *, minimum: int, maximum: int) -> int: if isinstance(value, bool) or not isinstance(value, int): raise ValidationError(f"{field_name} must be an integer") if value < minimum or value > maximum: raise ValidationError(f"{field_name} must be between {minimum} and {maximum}") return value def _require_optional_decimal(value: Any, field_name: str, *, minimum: Decimal, maximum: Decimal) -> Optional[Decimal]: if value is None: return None if isinstance(value, bool): raise ValidationError(f"{field_name} must be a decimal number") try: decimal_value = Decimal(str(value)) except Exception as exc: # pragma: no cover - defensive conversion raise ValidationError(f"{field_name} must be a decimal number") from exc if decimal_value < minimum or decimal_value > maximum: raise ValidationError(f"{field_name} must be between {minimum} and {maximum}") return decimal_value def _require_uuid(value: Any, field_name: str) -> UUID: if not isinstance(value, str): raise ValidationError(f"{field_name} must be a UUID string") try: return UUID(value) except Exception as exc: raise ValidationError(f"{field_name} must be a valid UUID") from exc def _require_iso_datetime(value: Any, field_name: str) -> datetime: if not isinstance(value, str): raise ValidationError(f"{field_name} must be an ISO 8601 datetime string") try: dt = datetime.fromisoformat(value.replace("Z", "+00:00")) except ValueError as exc: raise ValidationError(f"{field_name} must be a valid ISO 8601 datetime") from exc if dt.tzinfo is None: dt = dt.replace(tzinfo=timezone.utc) return dt def _require_string_list(value: Any, field_name: str, *, max_items: int = 20) -> list[str]: if not isinstance(value, list): raise ValidationError(f"{field_name} must be an array of strings") if len(value) > max_items: raise ValidationError(f"{field_name} must contain at most {max_items} items") result: list[str] = [] for index, item in enumerate(value): if not isinstance(item, str): raise ValidationError(f"{field_name}[{index}] must be a string") cleaned = item.strip() if not cleaned: raise ValidationError(f"{field_name}[{index}] must not be empty") result.append(cleaned) return result def validate_user_profile_update(payload: dict[str, Any]) -> UserProfileUpdate: """Convert untrusted JSON-like data into a validated typed object.""" data = _require_dict(payload) required_fields = {"user_id", "display_name", "requested_at"} missing = sorted(required_fields - set(data.keys())) if missing: raise ValidationError(f"Missing required field(s): {', '.join(missing)}") # Reject unexpected fields to reduce ambiguity and accidental drift. allowed_fields = { "user_id", "display_name", "age", "tags", "timezone_name", "credit_limit", "requested_at", } unexpected = sorted(set(data.keys()) - allowed_fields) if unexpected: raise ValidationError(f"Unexpected field(s): {', '.join(unexpected)}") user_id = _require_uuid(data["user_id"], "user_id") display_name = _require_string(data["display_name"], "display_name", min_len=1, max_len=80) age = None if "age" not in data else _require_int(data["age"], "age", minimum=0, maximum=150) tags = [] if "tags" not in data else _require_string_list(data["tags"], "tags", max_items=10) timezone_name = None if "timezone_name" not in data else _require_optional_string(data["timezone_name"], "timezone_name", max_len=64) credit_limit = None if "credit_limit" not in data else _require_optional_decimal( data["credit_limit"], "credit_limit", minimum=Decimal("0"), maximum=Decimal("1000000") ) requested_at = _require_iso_datetime(data["requested_at"], "requested_at") return UserProfileUpdate( user_id=user_id, display_name=display_name, age=age, tags=tags, timezone_name=timezone_name, credit_limit=credit_limit, requested_at=requested_at, ) def process_profile_update(update: UserProfileUpdate) -> dict[str, Any]: """Placeholder domain logic that only accepts already validated data.""" # In a real service, apply business rules here after authentication, # authorization, and validation have already been completed. return { "status": "accepted", "user_id": str(update.user_id), "display_name": update.display_name, "age": update.age, "tags": update.tags, "timezone_name": update.timezone_name, "credit_limit": str(update.credit_limit) if update.credit_limit is not None else None, "requested_at": update.requested_at.isoformat(), } def load_json_file(path: str) -> Any: with open(path, "r", encoding="utf-8") as handle: return json.load(handle) def main() -> int: parser = argparse.ArgumentParser( description="Validate a JSON payload using type hints plus runtime checks." ) parser.add_argument( "--input", required=True, help="Path to a JSON file containing the raw request payload.", ) parser.add_argument( "--pretty", action="store_true", help="Pretty-print the validated result.", ) args = parser.parse_args() try: raw_payload = load_json_file(args.input) validated = validate_user_profile_update(raw_payload) result = process_profile_update(validated) except FileNotFoundError: print(f"Error: input file not found: {args.input}", file=sys.stderr) return 2 except json.JSONDecodeError as exc: print(f"Error: invalid JSON: {exc}", file=sys.stderr) return 2 except ValidationError as exc: print(f"Validation error: {exc}", file=sys.stderr) return 3 if args.pretty: print(json.dumps(result, indent=2, sort_keys=True)) else: print(json.dumps(result, separators=(",", ":"), sort_keys=True)) return 0 if __name__ == "__main__": raise SystemExit(main())