Authorization

services:
  authorizer:
    configMap:
      authorizer:
        type: "Union"    # or "UserClouds" (legacy name, same engine)

services:
  authorizer:
    configMap:
      authorizer:
        type: "Union"
        bootstrap:
          organization: "<your-org-id>"
          domains:
            - development
            - staging
            - production
          serviceAccounts:
            - clientId: "<service-to-service-subject>"   # App 3 — sub claim value
              name: "service-to-service"
              role: "Admin"
            - clientId: "<operator-subject>"              # App 4 — sub claim value
              name: "operator"
              role: "Admin"
            - clientId: "<eager-subject>"                 # App 5 — sub claim value
              name: "eager"
              role: "Admin"
          adminUsers:
            - "<admin-email-or-subject>"

configMap:
  union:
    connection:
      trustedIdentityClaims:
        enabled: true
        externalIdentityClaim: ""           # The subject value of the internal S2S client
        externalIdentityTypeClaim: "app"    # Identity type for internal services

service AuthorizerService {
  rpc Authorize(AuthorizeRequest) returns (AuthorizeResponse);
}

port: 50051

# Internal platform service accounts — REQUIRED.
# Use the OAuth app client IDs from your identity provider (the same
# values configured during authentication setup). The server grants
# each the minimum permissions needed for the platform to function.
service_accounts:
  internal: "<INTERNAL_CLIENT_ID>"   # App 3: service-to-service (all actions)
  operator: "<AUTH_CLIENT_ID>"       # App 4: dataplane operator (cluster mgmt)
  eager: "<EAGER_CLIENT_ID>"         # App 5: task execution (register + launch)

#!/usr/bin/env python3
"""External authorization server for Union selfhosted deployments.

Demonstrates how to:
1. Extract identity and token from the Authorize() request
2. Grant required permissions to internal platform service accounts
3. Delegate human user authorization to your own logic
"""

import argparse
import base64
import json
import logging
from concurrent import futures

import grpc
import yaml

from gen.authorizer import authorizer_pb2_grpc, payload_pb2
from gen.common import authorization_pb2

log = logging.getLogger("authz")

# Action enum → name for logging
ACTION_NAMES = {
    v.number: v.name
    for v in authorization_pb2.DESCRIPTOR.enum_types_by_name["Action"].values
    if v.number != 0
}

# Minimum permissions each internal service account needs.
# These are non-negotiable — without them the platform breaks.
INTERNAL_SERVICE_ACCOUNT_PERMISSIONS = {
    # App 3 (INTERNAL_CLIENT_ID): the platform's own identity.
    # Used for all internal service-to-service communication.
    "internal": set(ACTION_NAMES.keys()),  # all actions
    # App 4 (AUTH_CLIENT_ID): dataplane operator.
    # Registers clusters, sends heartbeats, reads inventory.
    "operator": {
        authorization_pb2.ACTION_MANAGE_CLUSTER,
        authorization_pb2.ACTION_VIEW_FLYTE_INVENTORY,
        authorization_pb2.ACTION_VIEW_FLYTE_EXECUTIONS,
        authorization_pb2.ACTION_CREATE_FLYTE_EXECUTIONS,
    },
    # App 5 (EAGER client ID): task execution runtime.
    # Launches child tasks, registers workflows on behalf of users.
    "eager": {
        authorization_pb2.ACTION_VIEW_FLYTE_INVENTORY,
        authorization_pb2.ACTION_VIEW_FLYTE_EXECUTIONS,
        authorization_pb2.ACTION_REGISTER_FLYTE_INVENTORY,
        authorization_pb2.ACTION_CREATE_FLYTE_EXECUTIONS,
        authorization_pb2.ACTION_EDIT_EXECUTION_RELATED_ATTRIBUTES,
        authorization_pb2.ACTION_EDIT_CLUSTER_RELATED_ATTRIBUTES,
    },
}


def decode_jwt_payload(token: str) -> dict | None:
    """Base64-decode the JWT payload (no signature verification needed —
    the token is pre-validated by the platform)."""
    parts = token.split(".")
    if len(parts) != 3:
        return None
    payload = parts[1] + "=" * ((-len(parts[1])) % 4)
    return json.loads(base64.urlsafe_b64decode(payload))


class AuthorizerServicer(authorizer_pb2_grpc.AuthorizerServiceServicer):

    def __init__(self, config: dict):
        # Map internal service account client IDs → their permission sets.
        # These come from config.yaml and match the OAuth app client IDs
        # configured during authentication setup.
        self.service_accounts: dict[str, set[int]] = {}
        for key, client_id in config.get("service_accounts", {}).items():
            perms = INTERNAL_SERVICE_ACCOUNT_PERMISSIONS.get(key)
            if perms and client_id:
                self.service_accounts[client_id] = perms

    def Authorize(self, request, context):
        # --- 1. Extract subject from the proto identity ---
        # The proto type varies by IdP configuration:
        #   - external_identity: IdP without identitytype claim (most selfhosted)
        #   - user_id: IdP with identitytype="user"
        #   - application_id: IdP with identitytype="app"
        # Extract the subject from whichever type arrives.
        identity = request.identity
        subject = ""
        if identity.HasField("external_identity"):
            subject = identity.external_identity.subject
        elif identity.HasField("user_id"):
            subject = identity.user_id.subject
        elif identity.HasField("application_id"):
            subject = identity.application_id.subject

        # --- 2. Extract token from gRPC metadata ---
        # The JWT is forwarded in the "authorization" metadata header.
        # Not every request carries a token — internal background workers
        # may call Authorize() without one.
        metadata = dict(context.invocation_metadata())
        auth_header = metadata.get("authorization", "")
        token = auth_header[7:] if auth_header.lower().startswith("bearer ") else ""

        # --- 3. Decode JWT claims (if token available) ---
        # The token is pre-validated upstream — no signature verification
        # needed. Decode to read claims: sub, email, groups, etc.
        claims = decode_jwt_payload(token) if token else None

        # --- 4. Extract the action and resource ---
        action = request.action
        action_name = ACTION_NAMES.get(action, str(action))
        organization = request.organization

        # Resource is a oneof: project, domain, organization, or cluster
        resource = request.resource
        resource_desc = ""
        if resource.HasField("project"):
            p = resource.project
            domain = p.domain.name if p.HasField("domain") else "?"
            resource_desc = f"{organization}/{domain}/{p.name}"
        elif resource.HasField("domain"):
            resource_desc = f"{organization}/{resource.domain.name}"
        elif resource.HasField("cluster"):
            resource_desc = f"{organization}/{resource.cluster.name}"
        else:
            resource_desc = organization

        # --- 5. Authorize internal service accounts ---
        # The platform does NOT bypass external authorization for its own
        # service accounts. Your server MUST grant them the required
        # permissions or the platform will stop functioning.
        if subject in self.service_accounts:
            allowed = action in self.service_accounts[subject]
            log.info(
                "%s sub=%s (service-account) action=%s resource=%s",
                "ALLOWED" if allowed else "DENIED", subject, action_name, resource_desc,
            )
            return payload_pb2.AuthorizeResponse(allowed=allowed)

        # --- 6. Authorize human users ---
        # Replace this with your own authorization logic (OPA, Cedar,
        # internal RBAC, policy engine, token exchange, etc.)
        # Available data:
        #   - subject: the user's identity (from 'sub' claim)
        #   - claims: decoded JWT with sub, email, groups, identitytype, etc.
        #   - token: raw JWT for downstream use (may be empty for background workers)
        #   - action: the Action enum value being requested
        #   - resource: the target resource (project, domain, cluster, org)
        log.info("subject=%s action=%s resource=%s", subject, action_name, resource_desc)
        if claims:
            log.info(
                "  JWT: sub=%s type=%s iss=%s exp=%s",
                claims.get("sub"), claims.get("identitytype"),
                claims.get("iss"), claims.get("exp"),
            )

        allowed = True  # TODO: implement your authorization logic
        return payload_pb2.AuthorizeResponse(allowed=allowed)


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
    parser = argparse.ArgumentParser()
    parser.add_argument("--config", default="config.yaml")
    args = parser.parse_args()
    with open(args.config) as f:
        config = yaml.safe_load(f)

    server = grpc.server(futures.ThreadPoolExecutor(max_workers=4))
    authorizer_pb2_grpc.add_AuthorizerServiceServicer_to_server(
        AuthorizerServicer(config), server
    )
    port = config.get("port", 50051)
    server.add_insecure_port(f"[::]:{port}")
    server.start()
    log.info("External AuthZ server listening on port %d", port)
    server.wait_for_termination()

pip install grpcio grpcio-tools protobuf pyyaml
buf generate  # using buf.gen.yaml pointing to Union's IDL
python server.py --config config.yaml

kubectl get pods -n <controlplane-namespace> -l app.kubernetes.io/name=authorizer

kubectl logs -n <controlplane-namespace> deployment/authorizer | grep "Authz client config"
# Expected: Authz client config: type=External (or Noop, UserClouds)

kubectl logs -n <controlplane-namespace> deployment/authorizer | grep "external authorization"
# Expected: Initializing an external authorization proxy service with endpoint ...

uctl get project
uctl create execution --project <project> --domain development --launch-plan <launch-plan>

kubectl logs -n <controlplane-namespace> deployment/authorizer | grep "MANAGE_CLUSTER"

kubectl logs -n <controlplane-namespace> deployment/authorizer | grep "<EAGER_CLIENT_ID>"