Move auth related schema to inside the auth module (#1046)

Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>
Co-authored-by: Peyton Murray <[email protected]>

conda-store-server/conda_store_server/_internal/environment.py

@@ -10,6 +10,7 @@
 from sqlalchemy.orm import Query
 
 from conda_store_server._internal import conda_utils, orm, schema, utils
+from conda_store_server.server import schema as auth_schema
 
 
 def validate_environment(specification):
@@ -151,15 +152,15 @@ def _append_pip_packages(specification, packages):
 
 def filter_environments(
     query: Query,
-    role_bindings: schema.RoleBindings,
+    role_bindings: auth_schema.RoleBindings,
 ) -> Query:
     """Filter a query containing environments and namespaces by a set of role bindings.
 
     Parameters
     ----------
     query : Query
         Query containing both environments and namespaces
-    role_bindings : schema.RoleBindings
+    role_bindings : auth_schema.RoleBindings
         Role bindings to filter the results by
 
     Returns
@@ -171,7 +172,7 @@ def filter_environments(
     cases = []
     for entity_arn, entity_roles in role_bindings.items():
         namespace, name = utils.compile_arn_sql_like(
-            entity_arn, schema.ARN_ALLOWED_REGEX
+            entity_arn, auth_schema.ARN_ALLOWED_REGEX
         )
         cases.append(
             and_(

conda-store-server/conda_store_server/_internal/orm.py

@@ -42,6 +42,7 @@
 from conda_store_server._internal import conda_utils, schema, utils
 from conda_store_server._internal.environment import validate_environment
 from conda_store_server.exception import BuildPathError
+from conda_store_server.server import schema as auth_schema
 
 logger = logging.getLogger("orm")
 
@@ -101,7 +102,7 @@ class NamespaceRoleMapping(Base):
 
     @validates("entity")
     def validate_entity(self, key, entity):
-        if not schema.ARN_ALLOWED_REGEX.match(entity):
+        if not auth_schema.ARN_ALLOWED_REGEX.match(entity):
             raise ValueError(f"invalid entity={entity}")
 
         return entity

conda-store-server/conda_store_server/_internal/schema.py

@@ -4,11 +4,10 @@
 
 import datetime
 import enum
-import functools
 import os
 import re
 import sys
-from typing import Annotated, Any, Dict, List, Optional, TypeAlias, Union
+from typing import Annotated, Any, Dict, List, Optional, Union
 
 from conda_lock.lockfile.v1.models import Lockfile
 from pydantic import (
@@ -23,59 +22,7 @@
 from conda_store_server._internal import conda_utils
 from conda_store_server.exception import CondaStoreError
 
-
-def _datetime_factory(offset: datetime.timedelta):
-    """Utcnow datetime + timezone as string"""
-    return datetime.datetime.utcnow() + offset
-
-
-# An ARN is a string which matches namespaces and environments. For example:
-#     */*          matches all environments
-#     */team       matches all environments named 'team' in any namespace
-#
-# Namespaces and environment names cannot contain "*" ":" "#" " " "/"
 ALLOWED_CHARACTERS = "A-Za-z0-9-+_@$&?^~.="
-ARN_ALLOWED = f"^([{ALLOWED_CHARACTERS}*]+)/([{ALLOWED_CHARACTERS}*]+)$"
-ARN_ALLOWED_REGEX = re.compile(ARN_ALLOWED)
-
-
-#########################
-# Authentication Schema
-#########################
-
-RoleBindings: TypeAlias = Dict[
-    Annotated[str, StringConstraints(pattern=ARN_ALLOWED)], List[str]
-]
-
-
-class Permissions(enum.Enum):
-    """Permissions map to conda-store actions"""
-
-    ENVIRONMENT_CREATE = "environment:create"
-    ENVIRONMENT_READ = "environment::read"
-    ENVIRONMENT_UPDATE = "environment::update"
-    ENVIRONMENT_DELETE = "environment::delete"
-    ENVIRONMENT_SOLVE = "environment::solve"
-    BUILD_CANCEL = "build::cancel"
-    BUILD_DELETE = "build::delete"
-    NAMESPACE_CREATE = "namespace::create"
-    NAMESPACE_READ = "namespace::read"
-    NAMESPACE_UPDATE = "namespace::update"
-    NAMESPACE_DELETE = "namespace::delete"
-    NAMESPACE_ROLE_MAPPING_CREATE = "namespace-role-mapping::create"
-    NAMESPACE_ROLE_MAPPING_READ = "namespace-role-mapping::read"
-    NAMESPACE_ROLE_MAPPING_UPDATE = "namespace-role-mapping::update"
-    NAMESPACE_ROLE_MAPPING_DELETE = "namespace-role-mapping::delete"
-    SETTING_READ = "setting::read"
-    SETTING_UPDATE = "setting::update"
-
-
-class AuthenticationToken(BaseModel):
-    exp: datetime.datetime = Field(
-        default_factory=functools.partial(_datetime_factory, datetime.timedelta(days=1))
-    )
-    primary_namespace: str = "default"
-    role_bindings: RoleBindings = {}
 
 
 ##########################

conda-store-server/conda_store_server/_internal/server/views/api.py

@@ -14,11 +14,12 @@
 from conda_store_server import __version__, api
 from conda_store_server._internal import orm, schema
 from conda_store_server._internal.environment import filter_environments
-from conda_store_server._internal.schema import AuthenticationToken, Permissions
 from conda_store_server._internal.server import dependencies
 from conda_store_server.conda_store import CondaStore
 from conda_store_server.exception import CondaStoreError
+from conda_store_server.server import schema as auth_schema
 from conda_store_server.server.auth import Authentication
+from conda_store_server.server.schema import AuthenticationToken, Permissions
 
 
 class PaginatedArgs(TypedDict):
@@ -225,14 +226,14 @@ async def api_post_token(
     entity=Depends(dependencies.get_entity),
 ):
     if entity is None:
-        entity = schema.AuthenticationToken(
+        entity = auth_schema.AuthenticationToken(
             exp=datetime.datetime.now(tz=datetime.timezone.utc)
             + datetime.timedelta(days=1),
             primary_namespace=conda_store.config.default_namespace,
             role_bindings={},
         )
 
-    new_entity = schema.AuthenticationToken(
+    new_entity = auth_schema.AuthenticationToken(
         exp=expiration or entity.exp,
         primary_namespace=primary_namespace or entity.primary_namespace,
         role_bindings=role_bindings or auth.authorization.get_entity_bindings(entity),
@@ -672,7 +673,7 @@ async def api_list_environments(
         If specified, filter by environments containing the given package name(s)
     artifact : Optional[schema.BuildArtifactType]
         If specified, filter by environments with the given BuildArtifactType
-    jwt : Optional[schema.AuthenticationToken]
+    jwt : Optional[auth_schema.AuthenticationToken]
         If specified, retrieve only the environments accessible to this token; that is,
         only return environments that the user has 'admin', 'editor', and 'viewer'
         role bindings for.

conda-store-server/conda_store_server/_internal/server/views/registry.py

@@ -10,8 +10,8 @@
 
 from conda_store_server import api
 from conda_store_server._internal import orm, schema
-from conda_store_server._internal.schema import Permissions
 from conda_store_server._internal.server import dependencies
+from conda_store_server.server.schema import Permissions
 
 router_registry = APIRouter(tags=["registry"])
 

conda-store-server/conda_store_server/_internal/server/views/ui.py

@@ -12,8 +12,8 @@
 from conda_store_server._internal.action.generate_constructor_installer import (
     get_installer_platform,
 )
-from conda_store_server._internal.schema import Permissions
 from conda_store_server._internal.server import dependencies
+from conda_store_server.server.schema import Permissions
 
 router_ui = APIRouter(tags=["ui"])
 

conda-store-server/conda_store_server/api.py

@@ -11,6 +11,7 @@
 
 from conda_store_server._internal import conda_utils, orm, schema, utils
 from conda_store_server._internal.environment import filter_environments
+from conda_store_server.server import schema as auth_schema
 
 
 def list_namespaces(db, show_soft_deleted: bool = False):
@@ -284,7 +285,7 @@ def list_environments(
     artifact: schema.BuildArtifactType = None,
     search: str = None,
     show_soft_deleted: bool = False,
-    role_bindings: schema.RoleBindings | None = None,
+    role_bindings: auth_schema.RoleBindings | None = None,
 ) -> Query:
     """Retrieve all environments managed by conda-store.
 
@@ -308,7 +309,7 @@ def list_environments(
     show_soft_deleted : bool
         If specified, filter by environments which have a null value for the
         deleted_on attribute
-    role_bindings : schema.RoleBindings | None
+    role_bindings : auth_schema.RoleBindings | None
         If specified, filter by only the environments the given role_bindings
         have read, write, or admin access to. This should be the same object as
         the role bindings in conda_store_config.py, for example:

conda-store-server/conda_store_server/conda_store.py

@@ -18,6 +18,7 @@
 from conda_store_server.exception import CondaStoreError
 from conda_store_server.plugins import hookspec, plugin_manager
 from conda_store_server.plugins.types import lock
+from conda_store_server.server import schema as auth_schema
 
 
 class CondaStore:
@@ -258,7 +259,7 @@ def register_solve(self, db: Session, specification: schema.CondaSpecification):
             db=db,
             conda_store=self,
             namespace="solve",
-            action=schema.Permissions.ENVIRONMENT_SOLVE,
+            action=auth_schema.Permissions.ENVIRONMENT_SOLVE,
         )
 
         specification_model = self.config.validate_specification(
@@ -303,7 +304,7 @@ def register_environment(
             db=db,
             conda_store=self,
             namespace=namespace.name,
-            action=schema.Permissions.ENVIRONMENT_CREATE,
+            action=auth_schema.Permissions.ENVIRONMENT_CREATE,
         )
 
         if is_lockfile:
@@ -359,7 +360,7 @@ def create_build(self, db: Session, environment_id: int, specification_sha256: s
             db=db,
             conda_store=self,
             namespace=environment.namespace.name,
-            action=schema.Permissions.ENVIRONMENT_UPDATE,
+            action=auth_schema.Permissions.ENVIRONMENT_UPDATE,
         )
 
         settings = self.get_settings(
@@ -426,7 +427,7 @@ def update_environment_build(
             db=db,
             conda_store=self,
             namespace=namespace,
-            action=schema.Permissions.ENVIRONMENT_UPDATE,
+            action=auth_schema.Permissions.ENVIRONMENT_UPDATE,
         )
 
         build = api.get_build(db, build_id)
@@ -475,7 +476,7 @@ def delete_namespace(self, db: Session, namespace: str):
             db=db,
             conda_store=self,
             namespace=namespace,
-            action=schema.Permissions.NAMESPACE_DELETE,
+            action=auth_schema.Permissions.NAMESPACE_DELETE,
         )
 
         namespace = api.get_namespace(db, name=namespace)
@@ -502,7 +503,7 @@ def delete_environment(self, db: Session, namespace: str, name: str):
             db=db,
             conda_store=self,
             namespace=namespace,
-            action=schema.Permissions.ENVIRONMENT_DELETE,
+            action=auth_schema.Permissions.ENVIRONMENT_DELETE,
         )
 
         environment = api.get_environment(db, namespace=namespace, name=name)
@@ -531,7 +532,7 @@ def delete_build(self, db: Session, build_id: int):
             db=db,
             conda_store=self,
             namespace=build.environment.namespace.name,
-            action=schema.Permissions.BUILD_DELETE,
+            action=auth_schema.Permissions.BUILD_DELETE,
         )
 
         if build.status not in [

conda-store-server/conda_store_server/conda_store_config.py

@@ -21,6 +21,7 @@
 
 from conda_store_server import CONDA_STORE_DIR, BuildKey, api, storage
 from conda_store_server._internal import conda_utils, environment, schema, utils
+from conda_store_server.server import schema as auth_schema
 
 
 def conda_store_validate_specification(
@@ -48,14 +49,14 @@ def conda_store_validate_action(
     db: Session,
     conda_store,
     namespace: str,
-    action: schema.Permissions,
+    action: auth_schema.Permissions,
 ) -> None:
     settings = conda_store.get_settings(db)
     system_metrics = api.get_system_metrics(db)
 
     if action in (
-        schema.Permissions.ENVIRONMENT_CREATE,
-        schema.Permissions.ENVIRONMENT_UPDATE,
+        auth_schema.Permissions.ENVIRONMENT_CREATE,
+        auth_schema.Permissions.ENVIRONMENT_UPDATE,
     ) and (settings.storage_threshold > system_metrics.disk_free):
         raise utils.CondaStoreError(
             f"`CondaStore.storage_threshold` reached. Action {action.value} prevented due to insufficient storage space"