feat: add API key namespace restriction, MODULE_FULL type and g…

CHANGELOG.md

@@ -1,5 +1,22 @@
 # Changelog
 
+# [4.5.0](https://gitlab.dockstudios.co.uk/pub/terrareg/compare/v4.4.0...v4.5.0) (2026-05-12)
+
+
+### Features
+
+* **auth:** Add optional namespace restriction to upload and publish API keys
+* **auth:** Store matched DB-backed API key on Flask `g` during authentication to enable per-request namespace checks
+* **auth:** Add `MODULE_FULL` (`module_full`) API key type granting both upload and publish permissions in a single key
+* **ui:** Add Namespace Restriction field to the API Keys create form
+* **ui:** Add Namespace column to the API Keys list table
+
+
+### Notes
+
+* New nullable `namespace` column added to the `api_key` table — run migrations before deploying (`MIGRATE_DATABASE=True` or `alembic upgrade head`). Existing keys without a namespace set are unaffected.
+
+
 # [4.4.0](https://gitlab.dockstudios.co.uk/pub/terrareg/compare/v4.3.4...v4.4.0) (2026-05-10)
 
 

terrareg/alembic/versions/5aa1e8d0d9fb_add_api_key_table.py

@@ -0,0 +1,38 @@
+"""add_api_key_table
+
+Revision ID: 5aa1e8d0d9fb
+Revises: c72f7c6ef6a7
+Create Date: 2026-05-12 00:00:00.000000
+
+"""
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = '5aa1e8d0d9fb'
+down_revision = 'c72f7c6ef6a7'
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    op.create_table(
+        'api_key',
+        sa.Column('id', sa.Integer(), nullable=False, autoincrement=True),
+        sa.Column('name', sa.String(length=128), nullable=False),
+        sa.Column('key_type', sa.String(length=32), nullable=False),
+        sa.Column('key_prefix', sa.String(length=16), nullable=False),
+        sa.Column('key_hash', sa.String(length=128), nullable=False),
+        sa.Column('key_salt', sa.String(length=64), nullable=False),
+        sa.Column('created_at', sa.DateTime(), nullable=False),
+        sa.Column('created_by', sa.String(length=128), nullable=True),
+        sa.Column('last_used_at', sa.DateTime(), nullable=True),
+        sa.Column('expires_at', sa.DateTime(), nullable=True),
+        sa.Column('is_active', sa.Boolean(), nullable=False, server_default=sa.true()),
+        sa.PrimaryKeyConstraint('id')
+    )
+
+
+def downgrade():
+    op.drop_table('api_key')

terrareg/alembic/versions/a1b2c3d4e5f6_add_namespace_to_api_key.py

@@ -0,0 +1,24 @@
+"""add_namespace_to_api_key
+
+Revision ID: a1b2c3d4e5f6
+Revises: 5aa1e8d0d9fb
+Create Date: 2026-05-12 00:00:00.000000
+
+"""
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = 'a1b2c3d4e5f6'
+down_revision = '5aa1e8d0d9fb'
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    op.add_column('api_key', sa.Column('namespace', sa.String(length=128), nullable=True))
+
+
+def downgrade():
+    op.drop_column('api_key', 'namespace')

terrareg/auth/__init__.py

@@ -8,6 +8,7 @@
 from .admin_session_auth_method import AdminSessionAuthMethod
 from .upload_api_key_auth_method import UploadApiKeyAuthMethod
 from .publish_api_key_auth_method import PublishApiKeyAuthMethod
+from .module_full_api_key_auth_method import ModuleFullApiKeyAuthMethod
 from .saml_auth_method import SamlAuthMethod
 from .openid_auth_method import OpenidConnectAuthMethod
 from .github_auth_method import GithubAuthMethod
@@ -38,6 +39,7 @@ def get_current_auth_method(self) -> BaseAuthMethod:
                     AdminSessionAuthMethod,
                     UploadApiKeyAuthMethod,
                     PublishApiKeyAuthMethod,
+                    ModuleFullApiKeyAuthMethod,
                     SamlAuthMethod,
                     OpenidConnectAuthMethod,
                     GithubAuthMethod,

terrareg/auth/admin_api_key_auth_method.py

@@ -2,6 +2,7 @@
 from .base_admin_auth_method import BaseAdminAuthMethod
 from .base_api_key_auth_method import BaseApiKeyAuthMethod
 import terrareg.config
+import terrareg.models
 
 
 class AdminApiKeyAuthMethod(BaseAdminAuthMethod, BaseApiKeyAuthMethod):
@@ -10,4 +11,9 @@ class AdminApiKeyAuthMethod(BaseAdminAuthMethod, BaseApiKeyAuthMethod):
     @classmethod
     def check_auth_state(cls):
         """Check if admin API key is provided"""
-        return cls._check_api_key([terrareg.config.Config().ADMIN_AUTHENTICATION_TOKEN])
+        return cls._check_api_key([terrareg.config.Config().ADMIN_AUTHENTICATION_TOKEN], key_type=terrareg.models.ApiKeyType.ADMIN)
+
+    @classmethod
+    def is_enabled(cls):
+        """Whether admin API key auth is configured."""
+        return bool(terrareg.config.Config().ADMIN_AUTHENTICATION_TOKEN or terrareg.models.ApiKey.has_active_keys(terrareg.models.ApiKeyType.ADMIN))

terrareg/auth/base_api_key_auth_method.py

@@ -1,7 +1,10 @@
 
-from flask import request
+from flask import g, request
 
 from .base_auth_method import BaseAuthMethod
+import terrareg.models
+
+_MATCHED_API_KEY_G_KEY = '_matched_api_key'
 
 
 class BaseApiKeyAuthMethod(BaseAuthMethod):
@@ -12,11 +15,16 @@ def requires_csrf_tokens(self):
         """Whether auth type requires CSRF tokens"""
         return False
 
+    @property
+    def matched_api_key(self):
+        """Return the DB-backed ApiKey that authenticated this request, or None for env-var keys."""
+        return g.get(_MATCHED_API_KEY_G_KEY, None)
+
     @classmethod
-    def _check_api_key(cls, valid_keys):
+    def _check_api_key(cls, valid_keys, key_type=None):
         """Whether whether API key is valid"""
         if not isinstance(valid_keys, list):
-            return False
+            valid_keys = []
 
         # Obtain API key from request, ensuring that it is
         # not empty
@@ -29,6 +37,13 @@ def _check_api_key(cls, valid_keys):
             # Ensure the valid key is not empty:
             if actual_key and actual_key == valid_key:
                 return True
+
+        if key_type is not None:
+            stored_api_key = terrareg.models.ApiKey.verify_key(actual_key, key_type)
+            if stored_api_key is not None:
+                stored_api_key.mark_used()
+                setattr(g, _MATCHED_API_KEY_G_KEY, stored_api_key)
+                return True
         return False
 
     def can_access_read_api(self):

terrareg/auth/module_full_api_key_auth_method.py

@@ -0,0 +1,38 @@
+
+import terrareg.models
+from .base_api_key_auth_method import BaseApiKeyAuthMethod
+
+
+class ModuleFullApiKeyAuthMethod(BaseApiKeyAuthMethod):
+    """Auth method for module-full API key (upload + publish)"""
+
+    @classmethod
+    def check_auth_state(cls):
+        """Check if module-full API key is provided"""
+        return cls._check_api_key([], key_type=terrareg.models.ApiKeyType.MODULE_FULL)
+
+    @classmethod
+    def is_enabled(cls):
+        return terrareg.models.ApiKey.has_active_keys(terrareg.models.ApiKeyType.MODULE_FULL)
+
+    def can_upload_module_version(self, namespace):
+        """Whether user can upload/index module version within a namespace."""
+        key = self.matched_api_key
+        if key is not None and key.namespace is not None:
+            return key.namespace == namespace
+        return True
+
+    def can_publish_module_version(self, namespace):
+        """Whether user can publish module version within a namespace."""
+        key = self.matched_api_key
+        if key is not None and key.namespace is not None:
+            return key.namespace == namespace
+        return True
+
+    def check_namespace_access(self, permission_type, namespace):
+        """Check access level to a given namespace."""
+        return False
+
+    def get_username(self):
+        """Get username of current user"""
+        return 'Module Full API Key'

terrareg/auth/publish_api_key_auth_method.py

@@ -1,5 +1,6 @@
 
 import terrareg.config
+import terrareg.models
 from .base_api_key_auth_method import BaseApiKeyAuthMethod
 
 
@@ -9,14 +10,17 @@ class PublishApiKeyAuthMethod(BaseApiKeyAuthMethod):
     @classmethod
     def check_auth_state(cls):
         """Check if upload API key is provided"""
-        return cls._check_api_key(terrareg.config.Config().PUBLISH_API_KEYS)
+        return cls._check_api_key(terrareg.config.Config().PUBLISH_API_KEYS, key_type=terrareg.models.ApiKeyType.PUBLISH)
 
     @classmethod
     def is_enabled(cls):
-        return bool(terrareg.config.Config().PUBLISH_API_KEYS)
+        return bool(terrareg.config.Config().PUBLISH_API_KEYS or terrareg.models.ApiKey.has_active_keys(terrareg.models.ApiKeyType.PUBLISH))
 
     def can_publish_module_version(self, namespace):
         """Whether user can publish module version within a namespace."""
+        key = self.matched_api_key
+        if key is not None and key.namespace is not None:
+            return key.namespace == namespace
         return True
 
     def check_namespace_access(self, permission_type, namespace):

terrareg/auth/upload_api_key_auth_method.py

@@ -1,5 +1,6 @@
 
 import terrareg.config
+import terrareg.models
 from .base_api_key_auth_method import BaseApiKeyAuthMethod
 
 
@@ -9,14 +10,17 @@ class UploadApiKeyAuthMethod(BaseApiKeyAuthMethod):
     @classmethod
     def check_auth_state(cls):
         """Check if upload API key is provided"""
-        return cls._check_api_key(terrareg.config.Config().UPLOAD_API_KEYS)
+        return cls._check_api_key(terrareg.config.Config().UPLOAD_API_KEYS, key_type=terrareg.models.ApiKeyType.UPLOAD)
 
     @classmethod
     def is_enabled(cls):
-        return bool(terrareg.config.Config().UPLOAD_API_KEYS)
+        return bool(terrareg.config.Config().UPLOAD_API_KEYS or terrareg.models.ApiKey.has_active_keys(terrareg.models.ApiKeyType.UPLOAD))
 
     def can_upload_module_version(self, namespace):
         """Whether user can upload/index module version within a namespace."""
+        key = self.matched_api_key
+        if key is not None and key.namespace is not None:
+            return key.namespace == namespace
         return True
 
     def check_namespace_access(self, permission_type, namespace):

terrareg/database.py

@@ -56,6 +56,7 @@ def __init__(self):
         self._terraform_idp_authorization_code = None
         self._terraform_idp_access_token = None
         self._terraform_idp_subject_identifier = None
+        self._api_key = None
         self._user_group = None
         self._user_group_namespace_permission = None
         self._git_provider = None
@@ -108,6 +109,13 @@ def terraform_idp_subject_identifier(self):
             raise DatabaseMustBeIniistalisedError('Database class must be initialised.')
         return self._terraform_idp_subject_identifier
 
+    @property
+    def api_key(self):
+        """Return api_key table."""
+        if self._api_key is None:
+            raise DatabaseMustBeIniistalisedError('Database class must be initialised.')
+        return self._api_key
+
     @property
     def user_group(self):
         """Return user_group table."""
@@ -342,6 +350,22 @@ def initialise(self):
             sqlalchemy.Column('expiry', sqlalchemy.DateTime, nullable=False)
         )
 
+        self._api_key = sqlalchemy.Table(
+            'api_key', meta,
+            sqlalchemy.Column('id', sqlalchemy.Integer, primary_key=True, autoincrement=True),
+            sqlalchemy.Column('name', sqlalchemy.String(GENERAL_COLUMN_SIZE), nullable=False),
+            sqlalchemy.Column('key_type', sqlalchemy.String(32), nullable=False),
+            sqlalchemy.Column('key_prefix', sqlalchemy.String(16), nullable=False),
+            sqlalchemy.Column('key_hash', sqlalchemy.String(128), nullable=False),
+            sqlalchemy.Column('key_salt', sqlalchemy.String(64), nullable=False),
+            sqlalchemy.Column('created_at', sqlalchemy.DateTime, nullable=False),
+            sqlalchemy.Column('created_by', sqlalchemy.String(GENERAL_COLUMN_SIZE), nullable=True),
+            sqlalchemy.Column('last_used_at', sqlalchemy.DateTime, nullable=True),
+            sqlalchemy.Column('expires_at', sqlalchemy.DateTime, nullable=True),
+            sqlalchemy.Column('is_active', sqlalchemy.Boolean, nullable=False, default=True),
+            sqlalchemy.Column('namespace', sqlalchemy.String(GENERAL_COLUMN_SIZE), nullable=True),
+        )
+
         self._user_group = sqlalchemy.Table(
             'user_group', meta,
             sqlalchemy.Column('id', sqlalchemy.Integer, primary_key=True, autoincrement=True),

terrareg/errors.py

@@ -132,6 +132,24 @@ class InvalidGitProviderConfigError(TerraregError):
     pass
 
 
+class GitProviderInUseError(TerraregError):
+    """Git provider is in use by module providers and cannot be deleted."""
+
+    pass
+
+
+class GitProviderManagedByConfigurationError(TerraregError):
+    """Git provider is managed by configuration and cannot be edited via the UI."""
+
+    pass
+
+
+class InvalidApiKeyTypeError(TerraregError):
+    """API key type is invalid."""
+
+    pass
+
+
 class ModuleProviderCustomGitRepositoryUrlNotAllowedError(TerraregError):
     """Module provider cannot set custom git URL."""