""" Traffic generator for flask-demo-service and fastapi-demo-service. Sends a continuous stream of requests to both demo services using plain requests. Fetches a real Auth0 JWT once at startup and varies the Authorization header so MigrationAuthorizationBackend produces a realistic mix of metrics. Token distribution per request: 90% real JWT → real PDP decision 5% fake token → unauthenticated / pp_denied 5% no header → unauthenticated """ import os import random import sys import time import requests LOOP_INTERVAL = int(os.environ.get("LOOP_INTERVAL_SECONDS", "2")) REQUESTS_PER_LOOP = int(os.environ.get("REQUESTS_PER_LOOP", "5")) FLASK_BASE_URL = os.environ.get("FLASK_BASE_URL", "http://flask-demo:5000") FASTAPI_BASE_URL = os.environ.get("FASTAPI_BASE_URL", "http://fastapi-demo:8000") # Auth0 password-grant credentials (mirrors generate_bearer_jwt_token in jwtauth) AUTH0_URL = os.environ.get("AUTH0_URL", "https://qa-orchard.auth0.com/oauth/token") AUTH0_AUDIENCE = os.environ.get("AUTH0_AUDIENCE", "https://workstation.qaorch.com/api") AUTH0_USERNAME = os.environ.get("AUTH0_USERNAME") AUTH0_PASSWORD = os.environ.get("AUTH0_PASSWORD") AUTH0_CLIENT_ID = os.environ.get("AUTH0_CLIENT_ID") AUTH0_CLIENT_SECRET = os.environ.get("AUTH0_CLIENT_SECRET") PROFILE_TYPES = [ "LabelProfile", "ArtistProfile", "OrchAdminProfile", "PodcastProfile", "PublishingProfile", "InsightsProfile", "ContentProfile", "DistributionProfile", "Account360Profile", ] FLASK_ROUTES: list[tuple[str, str]] = [ ("GET", "/catalog/"), ("GET", "/catalog/1"), ("GET", "/catalog/42"), ("GET", "/catalog/100"), ("GET", "/content/1"), ("GET", "/content/42"), ("GET", "/content/100"), ("GET", "/release/1"), ("GET", "/release/42"), ("GET", "/release/100"), ("DELETE", "/release/1"), ("DELETE", "/release/42"), ] FASTAPI_ROUTES: list[tuple[str, str]] = [ ("GET", "/bulk-session/"), ("POST", "/bulk-session/"), ("GET", "/bulk-session/1"), ("GET", "/bulk-session/42"), ("GET", "/bulk-session/100"), ("PATCH", "/bulk-session/1"), ("DELETE", "/bulk-session/42"), ("GET", "/media/1"), ("GET", "/media/42"), ("GET", "/track/1"), ("GET", "/track/42"), ("DELETE", "/track/1"), ] _real_token: str | None = None def fetch_jwt() -> str | None: """Fetch a real Auth0 JWT using the password grant (once at startup). Mirrors jwtauth.testing.utils.generate_bearer_jwt_token. Returns None if credentials are missing or the request fails. """ if not all([AUTH0_USERNAME, AUTH0_PASSWORD, AUTH0_CLIENT_ID, AUTH0_CLIENT_SECRET]): print( "AUTH0_USERNAME/PASSWORD/CLIENT_ID/CLIENT_SECRET not set — " "running without a real token", flush=True, ) return None print(f"Fetching JWT for {AUTH0_USERNAME} ...", flush=True) try: resp = requests.post( AUTH0_URL, json={ "grant_type": "password", "username": AUTH0_USERNAME, "password": AUTH0_PASSWORD, "audience": AUTH0_AUDIENCE, "scope": "", "client_id": AUTH0_CLIENT_ID, "client_secret": AUTH0_CLIENT_SECRET, }, timeout=10, ) data = resp.json() except requests.RequestException as exc: print(f"Auth0 request failed: {exc} — running without a real token", file=sys.stderr, flush=True) return None except ValueError as exc: print(f"Auth0 response is not JSON: {exc} — running without a real token", file=sys.stderr, flush=True) return None if "error" in data: print(f"Auth0 error: {data}", file=sys.stderr, flush=True) return None token = data.get("access_token") if not token: print(f"Auth0 response missing access_token: {data}", file=sys.stderr, flush=True) return None expires_in = data.get("expires_in", "?") print(f"JWT obtained (expires_in={expires_in}s)", flush=True) return str(token) def pick_token() -> str | None: """Return a token variant: real (if available), fake, or None (no auth).""" choice = random.random() if choice < 0.05: return None # 5% — no auth header → unauthenticated elif choice < 0.10 or not _real_token: return "fake-token" # 5% — bad token → unauthenticated / pp_denied else: return _real_token # 90% — valid token → real PDP decision def pick_profile_headers() -> dict[str, str]: """Return a random set of Orchard profile headers, or an empty dict (~10% of the time).""" if random.random() < 0.10: return {} profile_type = random.choice(PROFILE_TYPES) return { "Orchard-Profile-Type": profile_type, "Orchard-Profile-Id": str(random.randint(100000, 999999)), "Orchard-Profile-UUID": f"{random.randint(0, 0xffffffff):08x}-{random.randint(0, 0xffff):04x}-" f"{random.randint(0, 0xffff):04x}-{random.randint(0, 0xffff):04x}-" f"{random.randint(0, 0xffffffffffff):012x}", } def send_one(label: str, base_url: str, routes: list[tuple[str, str]]) -> None: token = pick_token() method, path = random.choice(routes) url = base_url.rstrip("/") + path headers = {} if token: headers["Authorization"] = f"Bearer {token}" headers.update(pick_profile_headers()) response = requests.request(method, url, headers=headers, timeout=5) is_real = token == _real_token and _real_token is not None token_label = "real" if is_real else ("fake" if token else "none") profile_type = headers.get("Orchard-Profile-Type", "none") print( f"[{label}] {method} {path} [token={token_label}] [profile={profile_type}] → {response.status_code}", flush=True, ) def main() -> None: global _real_token _real_token = fetch_jwt() services = [ ("flask-demo-service", FLASK_BASE_URL, FLASK_ROUTES), ("fastapi-demo-service", FASTAPI_BASE_URL, FASTAPI_ROUTES), ] print( f"Sending {REQUESTS_PER_LOOP} requests every {LOOP_INTERVAL}s " f"to {FLASK_BASE_URL} and {FASTAPI_BASE_URL}", flush=True, ) while True: for _ in range(REQUESTS_PER_LOOP): label, base_url, routes = random.choice(services) try: send_one(label, base_url, routes) except Exception as e: print(f"Error [{label}]: {e}", flush=True) time.sleep(LOOP_INTERVAL) if __name__ == "__main__": main()