robots: Add robot federation for keyless auth (PROJQUAY-7803) (#3207)

robots: Add robot federation for keyless auth (PROJQUAY-7652)

adds the ability to configure federated auth for robots by
using external OIDC providers. Each robot can be configured
to have multiple external OIDC providers as the source for
authentication.

endpoints/api/robot.py

@@ -1,6 +1,9 @@
 """
 Manage user and organization robot accounts.
 """
+import json
+import logging
+
 from flask import abort, request
 
 from auth import scopes
@@ -9,7 +12,16 @@ from auth.permissions import (
     AdministerOrganizationPermission,
     OrganizationMemberPermission,
 )
+from data.database import FederatedLogin, LoginService
 from data.model import InvalidRobotException
+from data.model.user import (
+    attach_federated_login,
+    create_federated_user,
+    create_robot_federation_config,
+    delete_robot_federation_config,
+    get_robot_federation_config,
+    lookup_robot,
+)
 from endpoints.api import (
     ApiResource,
     allow_if_global_readonly_superuser,
@@ -48,8 +60,29 @@ CREATE_ROBOT_SCHEMA = {
     },
 }
 
+CREATE_ROBOT_FEDERATION_SCHEMA = {
+    "type": "array",
+    "description": "Federation configuration for the robot",
+    "items": {
+        "type": "object",
+        "properties": {
+            "issuer": {
+                "type": "string",
+                "description": "The issuer of the token",
+            },
+            "subject": {
+                "type": "string",
+                "description": "The subject of the token",
+            },
+        },
+        "required": ["issuer", "subject"],
+    },
+}
+
 ROBOT_MAX_SIZE = 1024 * 1024  # 1 KB.
 
+logger = logging.getLogger(__name__)
+
 
 def robots_list(prefix, include_permissions=False, include_token=False, limit=None):
     robots = model.list_entity_robot_permission_teams(
@@ -380,3 +413,81 @@ class RegenerateOrgRobot(ApiResource):
             return robot.to_dict(include_token=True)
 
         raise Unauthorized()
+
+
+# TODO: Add log action event for federation config changes
+@resource("/v1/organization/<orgname>/robots/<robot_shortname>/federation")
+@path_param("orgname", "The name of the organization")
+@path_param(
+    "robot_shortname", "The short name for the robot, without any user or organization prefix"
+)
+@related_user_resource(UserRobot)
+class OrgRobotFederation(ApiResource):
+
+    schemas = {
+        "CreateRobotFederation": CREATE_ROBOT_FEDERATION_SCHEMA,
+    }
+
+    @require_scope(scopes.ORG_ADMIN)
+    def get(self, orgname, robot_shortname):
+        permission = AdministerOrganizationPermission(orgname)
+        if permission.can() or allow_if_superuser() or allow_if_global_readonly_superuser():
+            robot_username = format_robot_username(orgname, robot_shortname)
+            robot = lookup_robot(robot_username)
+            return get_robot_federation_config(robot)
+
+        raise Unauthorized()
+
+    @require_scope(scopes.ORG_ADMIN)
+    @validate_json_request("CreateRobotFederation", optional=False)
+    def post(self, orgname, robot_shortname):
+        permission = AdministerOrganizationPermission(orgname)
+        if permission.can() or allow_if_superuser():
+            fed_config = self._parse_federation_config(request)
+
+            robot_username = format_robot_username(orgname, robot_shortname)
+            robot = lookup_robot(robot_username)
+            create_robot_federation_config(robot, fed_config)
+            log_action(
+                "create_robot_federation",
+                orgname,
+                {"config": fed_config, "robot": robot_shortname},
+            )
+            return fed_config
+
+        raise Unauthorized()
+
+    @require_scope(scopes.ORG_ADMIN)
+    def delete(self, orgname, robot_shortname):
+        permission = AdministerOrganizationPermission(orgname)
+        if permission.can() or allow_if_superuser():
+            robot_username = format_robot_username(orgname, robot_shortname)
+            robot = lookup_robot(robot_username)
+            delete_robot_federation_config(robot)
+            log_action(
+                "delete_robot_federation",
+                orgname,
+                {"robot": robot_shortname},
+            )
+            return "", 204
+        raise Unauthorized()
+
+    def _parse_federation_config(self, request):
+        fed_config = list()
+        seen = set()
+        for item in request.json:
+            if not item:
+                raise request_error(message="Missing one or more required fields (issuer, subject)")
+            issuer = item.get("issuer")
+            subject = item.get("subject")
+            if not issuer or not subject:
+                raise request_error(message="Missing one or more required fields (issuer, subject)")
+            entry = {"issuer": issuer, "subject": subject}
+
+            if f"{issuer}:{subject}" in seen:
+                raise request_error(message="Duplicate federation config entry")
+
+            seen.add(f"{issuer}:{subject}")
+            fed_config.append(entry)
+
+        return list(fed_config)