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 must 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,11 +34,44 @@ 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
 
+ @property
+ def is_supertoken(self) -> bool:
+ return self._is_supertoken
+
  def get_valid_api_key(self) -> Optional[ApiKey]:
  if not self.API_key:
  return None
@@ -135,6 +168,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()