import uuid from dataclasses import dataclass, field from typing import Any, Callable from ddtrace import tracer from flask import request from jwtauth import JWTAuth from jwtauth.utils import jwt_auth_from_environment from jwtauth.exceptions import JWTAuthError from oto import response from oto.adaptors.flask import flaskify from owsrequest import context as owsrequest_context, access, error_response, flask_request from python_pdp_sdk import ( ForwardKwargsGetter, UnauthenticatedException, UnauthorizedException, ) from sqlalchemy import text from product import config from product.api import migration_authorization_backend from product.connectors import mysql from product.constants import error _GET_TENANT_UUIDS = text(""" SELECT v.vendor_uuid, s.subaccount_uuid FROM vw_product p LEFT JOIN vendor v ON v.vendor_id = p.vendor_id LEFT JOIN subaccount s ON s.subaccount_id = p.subaccount_id WHERE p.release_id = :product_id """) @dataclass class Tenant: tenant_uuid: uuid.UUID tenant_type: str # "account" or "subaccount" tenant_attributes: dict = field(default_factory=dict) def only_for_identity(identity: str | list[str]) -> Callable: """Decorator function which checks if the identity of the user making the request matches the expected identities for the endpoint. .. code-block:: python @app.route("/") @only_for_identity("10436b38-5e11-472d-b6a4-bf1ee2b1b438") def index(): return "Hello, Ben!" @app.route("/two") @only_for_identity([ "10436b38-5e11-472d-b6a4-bf1ee2b1b438", "7f418dad-780b-4ab3-a4a2-2deba190f503" ]) def index(): return "Hello, Ben or Jess!" :param identity: The identity, or identities, of user(s) allowed to access the endpoint """ identity_list: list[str] = [identity] if isinstance(identity, str) else identity def decorator(function: Callable) -> Callable: auth: JWTAuth = jwt_auth_from_environment(environment=config.ENVIRONMENT) def wrapper(*args, **kwargs): try: context = owsrequest_context.get_request_context_from_headers( request.headers, jwt_auth_client=auth ) jwt_identity = context.jwt_identity_id if not jwt_identity: raise JWTAuthError("No valid identity found in request context") if jwt_identity not in identity_list: raise JWTAuthError( f"Identity {jwt_identity} is not permitted for this endpoint" ) except JWTAuthError as auth_error: return flaskify(response.Response(message=auth_error.message, status=403)) return function(*args, **kwargs) return wrapper return decorator def verify_profile_headers( profile_type: str, profile_id: str, required: bool = False, allowed_types: tuple[str, ...] | None = None, ) -> response.Response: """Verify profile headers presence and completeness. Args: profile_type: Profile type value. profile_id: Profile id value. required: If True, headers must be presented to pass the check. allowed_types: Allowed profile types to match the profile_type with. Returns: Response: Response 200 - if the check has passed, 404 - if headers are incomplete, 403 - if required headers are absent or profile_type doesn't match allowed_types. """ allowed_types = [_type.lower() for _type in allowed_types] if allowed_types else None completeness_response = access.verify_profile_headers_are_complete(profile_type, profile_id) if not completeness_response: return completeness_response if ( required and not profile_type or allowed_types and profile_type and profile_type.lower() not in allowed_types ): return error_response.create_error_forbidden() return response.Response(status=200) def get_tenant(product_id: int, subaccount_id: int | None) -> Tenant | None: """Return a Tenant built from the product’s vendor or subaccount UUID. Fetches UUIDs directly from the local DB via vw_product → vendor/subaccount joins. Prefers subaccount when subaccount_id is set. Note: tenant_hierarchy attributes (vendor_uuid, company_brand_uuid, parent_company_uuid) are not included here. If ows-pdp needs the full hierarchy, use ows-account instead: - POST /lookup/vendors/vendor-ids/ - POST /lookup/subaccounts/subaccount-ids/ Both endpoints return the full hierarchy and are used by ows-pdp’s own OwsAccountClient (see ows-pdp/pdp/connectors/ows_account.py). """ with mysql.db_session(read_only=True) as session: row = session.execute(_GET_TENANT_UUIDS, {"product_id": product_id}).fetchone() if not row: return None if subaccount_id and row.subaccount_uuid: return Tenant( tenant_uuid=uuid.UUID(row.subaccount_uuid), tenant_type="subaccount", ) if row.vendor_uuid: return Tenant( tenant_uuid=uuid.UUID(row.vendor_uuid), tenant_type="account", ) return None @tracer.wrap() def is_authorized_for_tenant( tenant_uuid: uuid.UUID, tenant_type: str = "account", resource_type: str = "digital_audio", action: str = "bulk_create", tenant_attributes: dict[str, Any] | None = None, ) -> bool: """Return True if the JWT identity is authorized for the tenant via ows-pdp. Mirrors is_authorized_for_tenant() in ows-product-staging/product_staging/api/auth.py. Propagates UnauthenticatedException — callers are responsible for mapping it to 401. """ return migration_authorization_backend.is_authorized( action=action, resource_id=0, resource_type=resource_type, resource_getter=ForwardKwargsGetter(), tenant={ **(tenant_attributes or {}), "tenant_type": tenant_type, "tenant_uuid": str(tenant_uuid), }, ) @tracer.wrap() def assert_authorization_for_tenant( tenant_uuid: uuid.UUID, tenant_type: str = "account", resource_type: str = "digital_audio", action: str = "bulk_create", tenant_attributes: dict[str, Any] | None = None, ) -> response.Response: """Imperative variant returning an oto `Response` for Flask handlers. Status semantics match the existing `verify_profile_headers` / `verify_grass_access` helpers so it slots into the established pattern: validation = auth.assert_authorization_for_tenant(vendor_uuid) if not validation: return flaskify(validation) 200 = allowed, 401 = no/invalid Authorization header, 403 = PDP denied. Rollout note: during the additive phase, callers may invoke this for observability while keeping the legacy access checks as the actual gate. Once every endpoint is covered we'll flip to using this as the sole gate and drop the legacy checks. """ try: authorized = is_authorized_for_tenant( tenant_uuid=tenant_uuid, tenant_type=tenant_type, resource_type=resource_type, action=action, tenant_attributes=tenant_attributes, ) except UnauthenticatedException: return response.create_error_response( error.ERROR_CODE_AUTHORIZATION, error.ERROR_MESSAGE_INVALID_JWT_NO_IDENTITY_ID, status=401, ) except UnauthorizedException: return error_response.create_error_forbidden() if not authorized: return error_response.create_error_forbidden() return response.Response(status=200) @tracer.wrap() def assert_authorization( *, product_id: int, vendor_id: int | None, subaccount_id: int | None, resource_type: str = "catalog", action: str = "read", ) -> bool: """Try PP authorization first; fall back to grass header check. Mirrors assert_authorization() in ows-product-staging/product_staging/logic/bulk_session.py but uses verify_grass_access instead of check_vendor_access_for_profiles. Returns True if authorized, False if denied. Raises python_pdp_sdk.UnauthenticatedException if the PP backend reports unauthenticated. Callers are responsible for mapping this to a 401 response. """ tenant = get_tenant(product_id=product_id, subaccount_id=subaccount_id) if tenant and is_authorized_for_tenant( tenant_uuid=tenant.tenant_uuid, tenant_type=tenant.tenant_type, resource_type=resource_type, action=action, tenant_attributes=tenant.tenant_attributes, ): return True return bool( flask_request.verify_grass_access( request, vendor=vendor_id, subaccount=subaccount_id, ) )