feat: Support pre-configured supertoken authentication

docs/source/deploying/configuration.rst

@@ -68,6 +68,10 @@ Configure default user permissions, creating default channel and super-admin per
 .. code::
 
    [users]
+   # an optional supertoken that can be used to bypass authorization
+   # e.g. for CI pipelines that need to rely on pre-configured tokens for the initial
+   # technial superuser. Length should be at least 32 characters.
+   supertoken = "use `openssl rand -hex 32` to generate a random token"
    # users with owner role
    admins = ["github:admin_user"]
    # users with maintainer role

quetz/authorization.py

@@ -34,8 +34,37 @@ class ServerRole(str, enum.Enum):
 
 
 class Rules:
-    def __init__(self, API_key: Optional[str], session: dict, db: Session):
-        self.API_key = API_key
+    def __init__(
+        self,
+        API_key: Optional[str],
+        session: dict,
+        db: Session,
+        supertoken: Optional[str] = None,
+    ):
+        """
+        Parameters
+        ----------
+        API_key: str
+            The API key used for authentication
+        session: dict
+            The session cookie used for authentication
+        db: Session
+            The database session
+        supertoken: str
+            Supertoken that can be used to bypass role checks
+        """
+        if supertoken and API_key == supertoken:
+            if len(supertoken) < 32:
+                raise HTTPException(
+                    status_code=status.HTTP_401_UNAUTHORIZED,
+                    detail="Supertoken too short, must be at least 32 characters long",
+                )
+            self.API_key = None
+            self.is_supertoken = True
+        else:
+            self.API_key = API_key
+            self.is_supertoken = False
+
         self.session = session
         self.db = db
 
@@ -135,6 +164,9 @@ def assert_assign_user_role(self, role: Optional[str]):
             return self.assert_server_roles([SERVER_OWNER, SERVER_MAINTAINER])
 
     def assert_server_roles(self, roles: list, msg: Optional[str] = None):
+        if self.is_supertoken:
+            return "supertoken"
+
         user_id = self.assert_user()
 
         if not self.has_server_roles(user_id, roles):

quetz/config.py

@@ -175,6 +175,7 @@ class Config:
         ConfigSection(
             "users",
             [
+                ConfigEntry("supertoken", str, default=None, required=False),
                 ConfigEntry("admins", list, default=list),
                 ConfigEntry("maintainers", list, default=list),
                 ConfigEntry("members", list, default=list),
@@ -495,6 +496,16 @@ def get_package_store(self) -> pkgstores.PackageStore:
                 }
             )
 
+    def get_supertoken(self) -> Optional[str]:
+        """Return the super token if it is set in the config.
+
+        Returns
+        -------
+        supertoken : Optional[str]
+            The super token
+        """
+        return self.config.get("users", {}).get("supertoken")
+
     def configured_section(self, section: str) -> bool:
         """Return if a given section has been configured.
 

quetz/deps.py

@@ -84,8 +84,14 @@ def get_rules(
     request: Request,
     session: dict = Depends(get_session),
     db: Session = Depends(get_db),
+    config: Config = Depends(get_config),
 ):
-    return authorization.Rules(request.headers.get("x-api-key"), session, db)
+    return authorization.Rules(
+        request.headers.get("x-api-key"),
+        session,
+        db,
+        supertoken=config.get_supertoken(),
+    )
 
 
 def get_tasks_worker(

quetz/tasks/workers.py

@@ -178,7 +178,12 @@ def job_wrapper(
         api_key = None
         if user_id:
             browser_session['user_id'] = user_id
-        auth = Rules(api_key, browser_session, db)
+        auth = Rules(
+            api_key,
+            browser_session,
+            db,
+            supertoken=config.get_supertoken(),
+        )
     if not session:
         session = get_remote_session()