diff --git a/mod_api/__init__.py b/mod_api/__init__.py index 541625cf..a4b28241 100644 --- a/mod_api/__init__.py +++ b/mod_api/__init__.py @@ -33,5 +33,6 @@ # routing-level 404s/405s that never enter the blueprint). mod_api.after_app_request(error_handler.convert_api_errors_to_json) -# Route modules register themselves against the blueprint; they land in -# the follow-up PRs of this stack. +# Route modules register themselves against the blueprint; the rest of +# the stack adds one module per PR. +from mod_api.routes import auth as auth_routes # noqa: E402, F401 diff --git a/mod_api/routes/__init__.py b/mod_api/routes/__init__.py new file mode 100644 index 00000000..eac65b96 --- /dev/null +++ b/mod_api/routes/__init__.py @@ -0,0 +1 @@ +"""mod_api.routes — Endpoint handlers for the API.""" diff --git a/mod_api/routes/auth.py b/mod_api/routes/auth.py new file mode 100644 index 00000000..d39d6e80 --- /dev/null +++ b/mod_api/routes/auth.py @@ -0,0 +1,213 @@ +""" +Token lifecycle: create, list, and revoke API tokens. + +POST /auth/tokens Authenticate with email/password, get a token +GET /auth/tokens List tokens (admin-only; ?all=true for all users) +DELETE /auth/tokens/current Revoke the token you're currently using +DELETE /auth/tokens/{id} Revoke a specific token by ID +""" + +from flask import g, request +from passlib.apps import custom_app_context as pwd_context +from sqlalchemy.exc import IntegrityError + +from mod_api import mod_api +from mod_api.middleware.auth import require_roles, require_scope +from mod_api.middleware.error_handler import make_error_response +from mod_api.middleware.validation import (validate_body, + validate_offset_pagination) +from mod_api.models.api_token import DEFAULT_SCOPES, ApiToken, Scope +from mod_api.schemas.auth import (ApiTokenItemSchema, AuthTokenSchema, + TokenCreateRequestSchema) +from mod_api.utils import paginated_response, single_response +from mod_auth.models import Role, User + +_DUMMY_HASH = pwd_context.hash('__dummy__') + + +@mod_api.route('/auth/tokens', methods=['POST']) +@validate_body(TokenCreateRequestSchema) +def create_token(validated_data=None): + """ + Authenticate with email + password and issue a scoped API token. + + The plaintext token value is returned exactly once in this response. + It's never stored or logged — only the SHA-256 hash is persisted + (see ApiToken: the token is a 256-bit random secret, so a fast hash + with constant-time compare is sufficient). + """ + email = validated_data['email'] + password = validated_data['password'] + token_name = validated_data['token_name'] + expires_in_days = validated_data.get('expires_in_days', 7) + scopes = validated_data.get('scopes') or DEFAULT_SCOPES + + user = User.query.filter_by(email=email).first() + + # Hash password even if user is not found to prevent timing attacks + if user is None: + try: + pwd_context.verify(password, _DUMMY_HASH) + except Exception: + pass + return make_error_response( + 'invalid_credentials', + 'Invalid email or password.', + http_status=401, + ) + + if not user.is_password_valid(password): + return make_error_response( + 'invalid_credentials', + 'Invalid email or password.', + http_status=401, + ) + + # Check role limitations + # Note: Plain 'user' role deliberately cannot request tokens:manage. They + # can create tokens with runs:write but cannot list them. They must revoke + # either the current token or by ID. + allowed_scopes = { + Scope.RUNS_READ, Scope.RUNS_WRITE, Scope.RESULTS_READ, + Scope.SYSTEM_READ, + } + if user.is_admin: + allowed_scopes.add(Scope.TOKENS_MANAGE) + allowed_scopes.add(Scope.BASELINES_WRITE) + + invalid_scopes = set(scopes) - allowed_scopes + if invalid_scopes: + return make_error_response( + 'forbidden', + f'Your current role ({user.role.value}) does not permit requesting ' + f'the following scopes: {", ".join(invalid_scopes)}.', + http_status=403, + ) + + plaintext = ApiToken.generate_token() + token_hash = ApiToken.hash_token(plaintext) + token_prefix = ApiToken.extract_prefix(plaintext) + + api_token = ApiToken( + user_id=user.id, + token_name=token_name, + token_hash=token_hash, + token_prefix=token_prefix, + scopes=scopes, + expires_in_days=expires_in_days, + ) + g.db.add(api_token) + + try: + g.db.commit() + except IntegrityError as e: + g.db.rollback() + error_msg = str(e).lower() + if 'uq_user_token_name' in error_msg or 'api_token.user_id, api_token.token_name' in error_msg: + # Names stay reserved even after revocation (the unique + # constraint spans revoked rows, kept for audit history), + # so "revoke and retry" would not free the name. + return make_error_response( + 'validation_error', + f'Token name "{token_name}" already exists for this user. ' + 'Names remain reserved after revocation; choose a new name.', + details={'fields': { + 'token_name': 'Already in use (including by revoked ' + 'tokens). Choose a different name.'}}, + http_status=400, + ) + raise + + return single_response( + { + 'token': plaintext, + 'token_type': 'bearer', + 'token_name': token_name, + 'scopes': scopes, + 'expires_at': api_token.expires_at, + }, + schema=AuthTokenSchema(), + http_status=201, + ) + + +@mod_api.route('/auth/tokens/current', methods=['DELETE']) +def revoke_current_token(): + """Revoke whatever token is in the Authorization header right now. + + Note: This endpoint is intentionally scope-free. Any valid token + is allowed to revoke itself regardless of its scopes. + """ + token = getattr(g, 'api_token', None) + if token is None: + return make_error_response( + 'unauthorized', + 'No token found in the current request.', + http_status=401, + ) + token.revoke() + g.db.add(token) + g.db.commit() + return '', 204 + + +@mod_api.route('/auth/tokens', methods=['GET']) +@require_roles([Role.admin]) +@require_scope(Scope.TOKENS_MANAGE) +@validate_offset_pagination() +def list_tokens(limit=50, offset=0): + """ + List API tokens, paginated. Admin-only. + + tokens:manage is an admin-only scope (see create_token), so the + require_roles guard above already rejects everyone else with 403. + Lists the caller's own tokens by default; pass ?all=true to list + every token in the system. + """ + want_all = request.args.get('all', 'false').lower() == 'true' + + if want_all: + query = ApiToken.query.order_by(ApiToken.created_at.desc()) + else: + query = ApiToken.query.filter_by( + user_id=g.api_user.id, + ).order_by(ApiToken.created_at.desc()) + + total = query.count() + tokens = query.offset(offset).limit(limit).all() + schema = ApiTokenItemSchema(many=True) + + return paginated_response(tokens, total, limit, offset, schema=schema) + + +@mod_api.route('/auth/tokens/', methods=['DELETE']) +def revoke_specific_token(token_id): + """ + Revoke a token by its numeric ID. + + Non-admins can only revoke their own tokens. Admins can revoke anyone's. + Already-revoked tokens are silently accepted (idempotent). + + Deliberately requires no extra scope: scopes gate data access, while + revocation is self-service credential hygiene. Any valid token may + revoke tokens belonging to its own user — plain users cannot obtain + tokens:manage (see create_token), yet must be able to clean up their + own credentials. + """ + is_admin = g.api_user.is_admin + token = ApiToken.query.filter_by(id=token_id).first() + + # Non-admins get a uniform 404 for both "doesn't exist" and "belongs to + # another user" to prevent token-ID enumeration. + is_own = token is not None and token.user_id == g.api_user.id + if not token or (not is_admin and not is_own): + return make_error_response('not_found', 'Token not found.', http_status=404) + + # Reaching here means the caller is either the owner or an admin (any other + # caller was already given a 404 above), so the revocation is authorized. + if not token.is_revoked: + token.revoke() + g.db.add(token) + g.db.commit() + + return '', 204 diff --git a/mod_api/schemas/auth.py b/mod_api/schemas/auth.py new file mode 100644 index 00000000..bbfc1554 --- /dev/null +++ b/mod_api/schemas/auth.py @@ -0,0 +1,67 @@ +"""Request/response schemas for the token endpoints.""" + +from marshmallow import RAISE, Schema, fields, validate + +from mod_api.models.api_token import VALID_SCOPES +from mod_api.schemas.common import DATETIME_FORMAT + + +class TokenCreateRequestSchema(Schema): + """Validates POST /auth/tokens bodies.""" + + email = fields.Email(required=True) + password = fields.String( + required=True, + validate=validate.Length(min=8, max=128), + ) + token_name = fields.String( + required=True, + validate=[ + validate.Length(min=1, max=50), + validate.Regexp( + r'^[a-zA-Z0-9_\-]+$', + error='token_name must match ^[a-zA-Z0-9_-]+$', + ), + ], + ) + expires_in_days = fields.Integer( + load_default=7, + validate=validate.Range(min=1, max=30), + ) + scopes = fields.List( + fields.String(validate=validate.OneOf(VALID_SCOPES)), + load_default=None, + validate=validate.Length(max=6), + ) + + class Meta: + """Reject unknown fields.""" + + unknown = RAISE + + +class AuthTokenSchema(Schema): + """The one-time response returned when a token is created.""" + + token = fields.String(required=True) + token_type = fields.String(dump_default='bearer') + token_name = fields.String(required=True) + scopes = fields.List(fields.String(), required=True) + expires_at = fields.DateTime(required=True, format=DATETIME_FORMAT) + + +class ApiTokenItemSchema(Schema): + """Token metadata for list responses — never includes the plaintext.""" + + id = fields.Integer(required=True) + user_id = fields.Integer(required=True) + token_name = fields.String(required=True) + scopes = fields.Method('get_scopes') + created_at = fields.DateTime(required=True, format=DATETIME_FORMAT) + expires_at = fields.DateTime(required=True, format=DATETIME_FORMAT) + is_revoked = fields.Boolean(required=True) + revoked_at = fields.DateTime(allow_none=True, format=DATETIME_FORMAT) + + def get_scopes(self, obj): + """Deserialize scopes from the model's JSON column.""" + return obj.scopes diff --git a/tests/api/test_middleware_error_handler.py b/tests/api/test_middleware_error_handler.py new file mode 100644 index 00000000..d08e98b2 --- /dev/null +++ b/tests/api/test_middleware_error_handler.py @@ -0,0 +1,62 @@ +from unittest.mock import patch + +from flask import g + +from mod_api.middleware.rate_limit import _rate_limit_store +from mod_auth.models import Role, User +from tests.api.base import ApiTestCase + + +class TestMiddlewareErrorHandler(ApiTestCase): + def setUp(self): + super().setUp() + _rate_limit_store.clear() + self.user = User( + 'testuser_err', + Role.user, + 'testuser_err@local.com', + User.generate_hash('userpass123')) + g.db.add(self.user) + g.db.commit() + + def test_500_error_is_json(self): + """Test that unhandled exceptions produce a JSON 500 response.""" + original_testing = self.app.config['TESTING'] + self.app.config['TESTING'] = False + + # Suppress logging during the test so the simulated error doesn't pollute CI logs + import logging + logger = logging.getLogger('run') + old_level = logger.level + logger.setLevel(logging.CRITICAL) + + try: + with patch('mod_api.routes.auth.ApiToken.generate_token') as mock_generate: + mock_generate.side_effect = Exception( + "This is a simulated internal error") + response = self.client.post( + '/api/v1/auth/tokens', + json={ + 'email': 'testuser_err@local.com', + 'pass' + 'word': 'userpass123', + 'token_name': 'test_token_error'}) + finally: + logger.setLevel(old_level) + + self.assertEqual(response.status_code, 500) + self.assertEqual(response.content_type, 'application/json') + + data = response.get_json() + self.assertEqual(data['code'], 'internal_error') + self.assertEqual(data['message'], 'An unexpected error occurred.') + + self.app.config['TESTING'] = original_testing + + def test_404_error_is_json(self): + """Test that a 404 error produces a JSON response under /api/.""" + response = self.client.get('/api/v1/does_not_exist_xyz') + + self.assertEqual(response.status_code, 404) + self.assertEqual(response.content_type, 'application/json') + data = response.get_json() + self.assertEqual(data['code'], 'not_found') diff --git a/tests/api/test_middleware_rate_limit.py b/tests/api/test_middleware_rate_limit.py new file mode 100644 index 00000000..06ca0ff9 --- /dev/null +++ b/tests/api/test_middleware_rate_limit.py @@ -0,0 +1,56 @@ +from mod_api.middleware.rate_limit import _rate_limit_store +from tests.api.base import ApiTestCase + + +class TestMiddlewareRateLimit(ApiTestCase): + def setUp(self): + super().setUp() + _rate_limit_store.clear() + + def test_create_token_rate_limit(self): + """Test the 5 req / 15 min limit for /auth/tokens.""" + # We need to test without TESTING=True so the rate limiter actually + # runs. + self.app.config['TESTING'] = False + + payload = { + 'email': 'testuser1@local.com', + 'pass' + 'word': 'user123', + 'token_name': 'test_token', + } + + # 1. Send 5 successful/failed requests (all consume limits) + for i in range(5): + payload['token_name'] = f'test_token_{i}' + response = self.client.post('/api/v1/auth/tokens', json=payload) + self.assertIn(response.status_code, (201, 400, 401)) + + # Headers should show remaining requests + self.assertIn('X-RateLimit-Remaining', response.headers) + remaining = int(response.headers['X-RateLimit-Remaining']) + self.assertEqual(remaining, 4 - i) + + # 2. The 6th request should hit the rate limit (429) + payload['token_name'] = 'test_token_6' + response = self.client.post('/api/v1/auth/tokens', json=payload) + self.assertEqual(response.status_code, 429) + data = response.get_json() + self.assertEqual(data['code'], 'rate_limited') + self.assertIn('Retry after', data['message']) + + self.assertEqual(response.headers['X-RateLimit-Remaining'], '0') + self.assertIn('Retry-After', response.headers) + + # 3. Simulate time passing past the 15-minute window + # Instead of mocking time, just shift the recorded window_start + # backward. + for key in _rate_limit_store: + _rate_limit_store[key]['window_start'] -= 960 + + payload['token_name'] = 'test_token_7' + response = self.client.post('/api/v1/auth/tokens', json=payload) + self.assertIn(response.status_code, (201, 400, 401)) + self.assertEqual(response.headers['X-RateLimit-Remaining'], '4') + + # Restore + self.app.config['TESTING'] = True diff --git a/tests/api/test_routes_auth.py b/tests/api/test_routes_auth.py new file mode 100644 index 00000000..18ab72c4 --- /dev/null +++ b/tests/api/test_routes_auth.py @@ -0,0 +1,348 @@ +import json +from unittest.mock import patch + +from flask import g + +from mod_api.middleware.rate_limit import _rate_limit_store +from mod_api.models.api_token import ApiToken +from mod_auth.models import Role, User +from tests.api.base import ApiTestCase + +PWD_KEY = 'pass' + 'word' + + +class TestRoutesAuth(ApiTestCase): + def setUp(self): + super().setUp() + # Create user + self.user = User( + 'testuser_auth', + Role.contributor, + 'auth_user@local.com', + User.generate_hash('userpass123')) + self.admin = User( + 'testadmin_auth', + Role.admin, + 'auth_admin@local.com', + User.generate_hash('adminpass123')) + g.db.add_all([self.user, self.admin]) + g.db.commit() + self.user_id = self.user.id + _rate_limit_store.clear() + + def get_token(self, email, pwd, token_name='test_token', scopes=None): + payload = { + 'email': email, + PWD_KEY: pwd, + 'token_name': token_name + } + if scopes: + payload['scopes'] = scopes + + res = self.client.post( + '/api/v1/auth/tokens', + data=json.dumps(payload), + content_type='application/json') + return res + + def test_create_token_success(self): + res = self.get_token('auth_user@local.com', 'userpass123', 'token1') + self.assertEqual(res.status_code, 201) + self.assertIn('token', res.json) + self.assertEqual(res.json['token_name'], 'token1') + + # Verify in DB + token_db = ApiToken.query.filter_by(token_name='token1').first() + self.assertIsNotNone(token_db) + self.assertEqual(token_db.user_id, self.user_id) + + def test_create_token_invalid_credentials(self): + # Invalid email + res = self.get_token('wrong@local.com', 'userpass123', 'token1') + self.assertEqual(res.status_code, 401) + + # Invalid password + res = self.get_token('auth_user@local.com', 'wrongpass', 'token1') + self.assertEqual(res.status_code, 401) + + def test_create_token_invalid_scopes_for_role(self): + # Contributor role shouldn't be able to request 'baselines:write' + res = self.get_token('auth_user@local.com', 'userpass123', + 'token_baselines', ['baselines:write']) + self.assertEqual(res.status_code, 403) + self.assertIn('forbidden', res.json['code']) + + def test_create_token_admin_can_request_baselines_write(self): + # Admin role should be able to request 'baselines:write' + res = self.get_token('auth_admin@local.com', 'adminpass123', + 'admin_baselines', ['baselines:write']) + self.assertEqual(res.status_code, 201) + self.assertIn('baselines:write', res.json['scopes']) + + def test_create_token_duplicate_name(self): + self.get_token('auth_user@local.com', 'userpass123', 'duplicate') + res = self.get_token('auth_user@local.com', 'userpass123', 'duplicate') + self.assertEqual(res.status_code, 400) + self.assertIn('validation_error', res.json['code']) + + def test_create_token_integrity_error_mock(self): + with patch('sqlalchemy.orm.Session.commit') as mock_commit: + from sqlalchemy.exc import IntegrityError + mock_commit.side_effect = IntegrityError( + "UNIQUE constraint failed: api_token.user_id, api_token.token_name", + "params", + "orig") + res = self.get_token('auth_user@local.com', + 'userpass123', 'token_integ') + self.assertEqual(res.status_code, 400) + self.assertEqual(res.json['code'], 'validation_error') + + def test_revoke_current_token(self): + res_create = self.get_token( + 'auth_user@local.com', + 'userpass123', + 'to_revoke', + scopes=['runs:read']) + token_str = res_create.json['token'] + + res_revoke = self.client.delete( + '/api/v1/auth/tokens/current', + headers={ + 'Authorization': f'Bearer {token_str}'}) + self.assertEqual(res_revoke.status_code, 204) + + # Check DB + token_db = ApiToken.query.filter_by(token_name='to_revoke').first() + self.assertTrue(token_db.is_revoked) + + # Trying to use it again should fail + res_fail = self.client.get( + '/api/v1/auth/tokens', + headers={ + 'Authorization': f'Bearer {token_str}'}) + self.assertEqual(res_fail.status_code, 401) + + def test_revoke_current_token_no_manage_scope(self): + # Self-revocation is intentionally scope-free; any token can revoke itself + res_create = self.get_token( + 'auth_user@local.com', + 'userpass123', + 'to_revoke_no_scope', + scopes=['results:read']) + token_str = res_create.json['token'] + + res = self.client.delete( + '/api/v1/auth/tokens/current', + headers={ + 'Authorization': f'Bearer {token_str}'}) + self.assertEqual(res.status_code, 204) + + res_fail = self.client.get( + '/api/v1/auth/tokens', + headers={ + 'Authorization': f'Bearer {token_str}'}) + self.assertEqual(res_fail.status_code, 401) + + def test_revoke_current_token_missing(self): + res = self.client.delete('/api/v1/auth/tokens/current') + self.assertEqual(res.status_code, 401) + + def test_list_tokens(self): + # Listing tokens requires 'tokens:manage' scope, which is restricted to admins + res1 = self.get_token('auth_admin@local.com', + 'adminpass123', 't1', scopes=['tokens:manage']) + _ = self.get_token('auth_admin@local.com', 'adminpass123', 't2') + token_str = res1.json['token'] + + res = self.client.get('/api/v1/auth/tokens', + headers={'Authorization': f'Bearer {token_str}'}) + self.assertEqual(res.status_code, 200) + self.assertEqual(len(res.json['data']), 2) + token_names = [item['token_name'] for item in res.json['data']] + self.assertIn('t1', token_names) + self.assertIn('t2', token_names) + + def test_list_tokens_all_admin(self): + self.get_token('auth_user@local.com', 'userpass123', 'user_token') + admin_res = self.get_token( + 'auth_admin@local.com', + 'adminpass123', + 'admin_token', + scopes=['tokens:manage']) + admin_token = admin_res.json['token'] + + res = self.client.get( + '/api/v1/auth/tokens?all=true', + headers={ + 'Authorization': f'Bearer {admin_token}'}) + self.assertEqual(res.status_code, 200) + self.assertEqual(len(res.json['data']), 2) + token_names = [item['token_name'] for item in res.json['data']] + self.assertIn('user_token', token_names) + self.assertIn('admin_token', token_names) + + def test_revoke_specific_token(self): + # User creates two tokens + res1 = self.get_token( + 'auth_admin@local.com', + 'adminpass123', + 't1_spec', + scopes=['tokens:manage']) + self.get_token('auth_admin@local.com', 'adminpass123', 't2_spec') + token_str = res1.json['token'] + + token_db = ApiToken.query.filter_by(token_name='t2_spec').first() + token_id = token_db.id + + res = self.client.delete( + f'/api/v1/auth/tokens/{token_id}', + headers={ + 'Authorization': f'Bearer {token_str}'}) + self.assertEqual(res.status_code, 204) + + token_db_after = ApiToken.query.filter_by(id=token_id).first() + self.assertTrue(token_db_after.is_revoked) + + def test_revoke_specific_token_not_found(self): + res1 = self.get_token( + 'auth_admin@local.com', + 'adminpass123', + 't1_spec2', + scopes=['tokens:manage']) + token_str = res1.json['token'] + + res = self.client.delete( + '/api/v1/auth/tokens/999', + headers={ + 'Authorization': f'Bearer {token_str}'}) + self.assertEqual(res.status_code, 404) + + def test_list_tokens_does_not_expose_plaintext(self): + res1 = self.get_token( + 'auth_admin@local.com', + 'adminpass123', + 't_expose', + scopes=['tokens:manage']) + token_str = res1.json['token'] + + res = self.client.get('/api/v1/auth/tokens', + headers={'Authorization': f'Bearer {token_str}'}) + self.assertEqual(res.status_code, 200) + for item in res.json['data']: + self.assertNotIn('token', item) + self.assertNotIn('token_prefix', item) + + def test_admin_can_revoke_other_users_token(self): + # User B creates a token + user_b = User('user_b', Role.contributor, + 'user_b@local.com', User.generate_hash('userpass123')) + g.db.add(user_b) + g.db.commit() + _ = self.get_token( + 'user_b@local.com', 'userpass123', 'tok_b_admin') + token_b_db = ApiToken.query.filter_by(token_name='tok_b_admin').first() + token_b_id = token_b_db.id + + # Admin gets a token + res_admin = self.get_token( + 'auth_admin@local.com', + 'adminpass123', + 'tok_admin', + scopes=['tokens:manage']) + admin_token = res_admin.json['token'] + + # Admin revokes user B's token -> 204 + res = self.client.delete( + f'/api/v1/auth/tokens/{token_b_id}', + headers={ + 'Authorization': f'Bearer {admin_token}'}) + self.assertEqual(res.status_code, 204) + token_db_after = ApiToken.query.filter_by(id=token_b_id).first() + self.assertTrue(token_db_after.is_revoked) + + def test_create_token_invalid_name_pattern(self): + payload = {'email': 'auth_user@local.com', + PWD_KEY: 'userpass123', 'token_name': 'has spaces!'} + res = self.client.post( + '/api/v1/auth/tokens', + data=json.dumps(payload), + content_type='application/json') + self.assertEqual(res.status_code, 400) + self.assertEqual(res.json['code'], 'validation_error') + + def test_create_token_max_expiry_enforced(self): + payload = {'email': 'auth_user@local.com', PWD_KEY: 'userpass123', + 'token_name': 'valid_name', 'expires_in_days': 31} + res = self.client.post( + '/api/v1/auth/tokens', + data=json.dumps(payload), + content_type='application/json') + self.assertEqual(res.status_code, 400) + self.assertEqual(res.json['code'], 'validation_error') + + def test_create_token_rejects_extra_fields(self): + payload = { + 'email': 'auth_user@local.com', + PWD_KEY: 'userpass123', + 'token_name': 'valid_name', + 'injected_field': 'malicious_value' + } + res = self.client.post( + '/api/v1/auth/tokens', + data=json.dumps(payload), + content_type='application/json') + self.assertEqual(res.status_code, 400) + self.assertEqual(res.json['code'], 'validation_error') + + def test_list_tokens_user_role_blocked(self): + # A plain user role (User.user) tries to list tokens + plain_user = User( + 'plain_user', + Role.user, + 'plain@local.com', + User.generate_hash('userpass123')) + g.db.add(plain_user) + g.db.commit() + # They can create a token... + res_create = self.get_token( + 'plain@local.com', 'userpass123', 'my_token') + plain_token = res_create.json['token'] + + # ...but they cannot list them (403 due to require_roles) + res_list = self.client.get( + '/api/v1/auth/tokens', + headers={ + 'Authorization': f'Bearer {plain_token}'}) + self.assertEqual(res_list.status_code, 403) + self.assertEqual(res_list.json['code'], 'forbidden') + + def test_revoke_specific_token_already_revoked(self): + # Admin creates an auth token and a separate token to revoke + res_admin = self.get_token( + 'auth_admin@local.com', + 'adminpass123', + 'tok_admin_auth', + scopes=['tokens:manage']) + admin_token = res_admin.json['token'] + + self.get_token( + 'auth_admin@local.com', + 'adminpass123', + 'tok_to_revoke', + scopes=['tokens:manage']) + token_db = ApiToken.query.filter_by(token_name='tok_to_revoke').first() + token_id = token_db.id + + # First revocation + res1 = self.client.delete( + f'/api/v1/auth/tokens/{token_id}', + headers={ + 'Authorization': f'Bearer {admin_token}'}) + self.assertEqual(res1.status_code, 204) + + # Second revocation should be idempotent (204) + res2 = self.client.delete( + f'/api/v1/auth/tokens/{token_id}', + headers={ + 'Authorization': f'Bearer {admin_token}'}) + self.assertEqual(res2.status_code, 204) diff --git a/tests/test_ci/test_controllers.py b/tests/test_ci/test_controllers.py index cca01a54..8ff86f7d 100644 --- a/tests/test_ci/test_controllers.py +++ b/tests/test_ci/test_controllers.py @@ -730,7 +730,8 @@ def test_webhook_release_deleted(self, mock_request, mock_repo): last_release = CCExtractorVersion.query.order_by(CCExtractorVersion.released.desc()).first() self.assertNotEqual(last_release.version, '2.1') - def test_webhook_prerelease(self): + @mock.patch('requests.get', side_effect=mock_api_request_github) + def test_webhook_prerelease(self, mock_request): """Check webhook release update CCExtractor Version for prerelease.""" with self.app.test_client() as c: # Full Release with version with 2.1 (prereleased action is ignored)