forked from 0xWheatyz/SPARC
Add user-level API key generation for programmatic access
- Add api_keys table (id, user_id, key_hash, label, created_at) to schema
- Add POST /auth/apikeys to generate 32-byte hex API keys (bcrypt-hashed)
- Add GET /auth/apikeys to list active key metadata (no secrets)
- Add DELETE /auth/apikeys/{key_id} to revoke keys
- Extend get_current_user to accept either JWT Bearer or X-API-Key header
- Plaintext key returned only at creation time
- 16 new tests covering creation, listing, revocation, auth, and full flow
Closes leeworks-agents/SPARC#1673
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,319 @@
|
||||
"""Tests for user-level API key generation, listing, revocation, and authentication.
|
||||
|
||||
Covers all acceptance criteria from issue #1673:
|
||||
1. Users can create API keys (POST /auth/apikeys)
|
||||
2. Users can list their active key IDs (GET /auth/apikeys)
|
||||
3. Users can revoke keys (DELETE /auth/apikeys/{key_id})
|
||||
4. API requests authenticated with a valid API key work on protected endpoints
|
||||
5. Revoked keys are immediately rejected
|
||||
6. Plaintext key is shown only at creation time
|
||||
|
||||
All tests use mocked DB fixtures and require no live database.
|
||||
"""
|
||||
|
||||
from datetime import datetime, timezone
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from SPARC.api import app
|
||||
from SPARC.auth import (
|
||||
create_access_token,
|
||||
generate_api_key,
|
||||
hash_api_key,
|
||||
verify_api_key,
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def client():
|
||||
"""Create test client."""
|
||||
return TestClient(app)
|
||||
|
||||
|
||||
def _make_user():
|
||||
return {
|
||||
"id": 1,
|
||||
"email": "user@test.com",
|
||||
"role": "user",
|
||||
"created_at": datetime(2025, 1, 1, tzinfo=timezone.utc),
|
||||
}
|
||||
|
||||
|
||||
def _auth_header(user_dict):
|
||||
"""Create an Authorization header with a valid access token."""
|
||||
token = create_access_token(user_dict["id"], user_dict["email"], user_dict["role"])
|
||||
return {"Authorization": f"Bearer {token}"}
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def mock_db(monkeypatch):
|
||||
"""Mock the database client used by auth and api endpoints."""
|
||||
db = MagicMock()
|
||||
db.get_user_count.return_value = 0
|
||||
db.get_user_by_id.return_value = None
|
||||
db.get_user_by_email.return_value = None
|
||||
db.authenticate_user.return_value = None
|
||||
db.create_user.return_value = None
|
||||
db.get_all_users.return_value = []
|
||||
db.update_user_role.return_value = None
|
||||
db.delete_user.return_value = False
|
||||
db.create_api_key.return_value = None
|
||||
db.list_api_keys.return_value = []
|
||||
db.delete_api_key.return_value = False
|
||||
db.get_all_api_key_hashes.return_value = []
|
||||
|
||||
with patch("SPARC.api.get_db_client", return_value=db), \
|
||||
patch("SPARC.auth.get_db_client", return_value=db):
|
||||
yield db
|
||||
|
||||
|
||||
class TestCreateApiKey:
|
||||
"""POST /auth/apikeys"""
|
||||
|
||||
def test_create_key_returns_plaintext_and_id(self, client, mock_db):
|
||||
"""Creating a key returns the plaintext key and metadata."""
|
||||
user = _make_user()
|
||||
mock_db.get_user_by_id.return_value = user
|
||||
mock_db.create_api_key.return_value = {
|
||||
"id": 42,
|
||||
"user_id": user["id"],
|
||||
"label": "my-ci-key",
|
||||
"created_at": datetime(2025, 6, 1, tzinfo=timezone.utc),
|
||||
}
|
||||
|
||||
response = client.post(
|
||||
"/auth/apikeys",
|
||||
json={"label": "my-ci-key"},
|
||||
headers=_auth_header(user),
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["id"] == 42
|
||||
assert len(data["key"]) == 64 # 32 bytes hex = 64 chars
|
||||
assert data["label"] == "my-ci-key"
|
||||
assert "created_at" in data
|
||||
|
||||
# Verify the hash passed to DB is valid for the returned key
|
||||
call_args = mock_db.create_api_key.call_args
|
||||
stored_hash = call_args.kwargs.get("key_hash") or call_args[1].get("key_hash") or call_args[0][1]
|
||||
assert verify_api_key(data["key"], stored_hash)
|
||||
|
||||
def test_create_key_without_label(self, client, mock_db):
|
||||
"""Creating a key without a label should work."""
|
||||
user = _make_user()
|
||||
mock_db.get_user_by_id.return_value = user
|
||||
mock_db.create_api_key.return_value = {
|
||||
"id": 1,
|
||||
"user_id": user["id"],
|
||||
"label": None,
|
||||
"created_at": datetime(2025, 6, 1, tzinfo=timezone.utc),
|
||||
}
|
||||
|
||||
response = client.post(
|
||||
"/auth/apikeys",
|
||||
headers=_auth_header(user),
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.json()["label"] is None
|
||||
|
||||
def test_create_key_requires_auth(self, client):
|
||||
"""Creating a key without auth should fail."""
|
||||
response = client.post("/auth/apikeys")
|
||||
assert response.status_code == 401
|
||||
|
||||
|
||||
class TestListApiKeys:
|
||||
"""GET /auth/apikeys"""
|
||||
|
||||
def test_list_keys_returns_metadata_only(self, client, mock_db):
|
||||
"""Listing keys should return IDs and labels, not secrets."""
|
||||
user = _make_user()
|
||||
mock_db.get_user_by_id.return_value = user
|
||||
mock_db.list_api_keys.return_value = [
|
||||
{"id": 1, "label": "key-1", "created_at": datetime(2025, 1, 1, tzinfo=timezone.utc)},
|
||||
{"id": 2, "label": None, "created_at": datetime(2025, 2, 1, tzinfo=timezone.utc)},
|
||||
]
|
||||
|
||||
response = client.get("/auth/apikeys", headers=_auth_header(user))
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert len(data) == 2
|
||||
assert data[0]["id"] == 1
|
||||
assert data[0]["label"] == "key-1"
|
||||
# Ensure no secret key is exposed
|
||||
for item in data:
|
||||
assert "key" not in item
|
||||
assert "key_hash" not in item
|
||||
|
||||
def test_list_keys_empty(self, client, mock_db):
|
||||
"""User with no keys gets an empty list."""
|
||||
user = _make_user()
|
||||
mock_db.get_user_by_id.return_value = user
|
||||
mock_db.list_api_keys.return_value = []
|
||||
|
||||
response = client.get("/auth/apikeys", headers=_auth_header(user))
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.json() == []
|
||||
|
||||
|
||||
class TestRevokeApiKey:
|
||||
"""DELETE /auth/apikeys/{key_id}"""
|
||||
|
||||
def test_revoke_existing_key(self, client, mock_db):
|
||||
"""Revoking an owned key should succeed."""
|
||||
user = _make_user()
|
||||
mock_db.get_user_by_id.return_value = user
|
||||
mock_db.delete_api_key.return_value = True
|
||||
|
||||
response = client.delete("/auth/apikeys/42", headers=_auth_header(user))
|
||||
|
||||
assert response.status_code == 200
|
||||
assert "revoked" in response.json()["message"].lower()
|
||||
mock_db.delete_api_key.assert_called_once_with(42, user["id"])
|
||||
|
||||
def test_revoke_nonexistent_key_returns_404(self, client, mock_db):
|
||||
"""Revoking a key that doesn't exist (or isn't owned) returns 404."""
|
||||
user = _make_user()
|
||||
mock_db.get_user_by_id.return_value = user
|
||||
mock_db.delete_api_key.return_value = False
|
||||
|
||||
response = client.delete("/auth/apikeys/999", headers=_auth_header(user))
|
||||
|
||||
assert response.status_code == 404
|
||||
|
||||
|
||||
class TestApiKeyAuthentication:
|
||||
"""Using X-API-Key header on protected endpoints."""
|
||||
|
||||
def test_valid_api_key_accesses_protected_endpoint(self, client, mock_db):
|
||||
"""A valid API key should authenticate and access /auth/me."""
|
||||
user = _make_user()
|
||||
plaintext = generate_api_key()
|
||||
hashed = hash_api_key(plaintext)
|
||||
|
||||
mock_db.get_all_api_key_hashes.return_value = [
|
||||
{"key_hash": hashed, "user_id": user["id"]},
|
||||
]
|
||||
mock_db.get_user_by_id.return_value = user
|
||||
|
||||
response = client.get("/auth/me", headers={"X-API-Key": plaintext})
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["email"] == user["email"]
|
||||
assert data["id"] == user["id"]
|
||||
|
||||
def test_invalid_api_key_returns_401(self, client, mock_db):
|
||||
"""An invalid API key should return 401."""
|
||||
mock_db.get_all_api_key_hashes.return_value = []
|
||||
|
||||
response = client.get("/auth/me", headers={"X-API-Key": "bad-key"})
|
||||
|
||||
assert response.status_code == 401
|
||||
assert "invalid api key" in response.json()["detail"].lower()
|
||||
|
||||
def test_revoked_key_returns_401(self, client, mock_db):
|
||||
"""After revocation, using the key should return 401."""
|
||||
# Simulate revoked key: no matching hashes in DB
|
||||
mock_db.get_all_api_key_hashes.return_value = []
|
||||
|
||||
response = client.get("/auth/me", headers={"X-API-Key": "a" * 64})
|
||||
|
||||
assert response.status_code == 401
|
||||
|
||||
def test_api_key_for_deleted_user_returns_401(self, client, mock_db):
|
||||
"""An API key whose user no longer exists should return 401."""
|
||||
plaintext = generate_api_key()
|
||||
hashed = hash_api_key(plaintext)
|
||||
|
||||
mock_db.get_all_api_key_hashes.return_value = [
|
||||
{"key_hash": hashed, "user_id": 999},
|
||||
]
|
||||
mock_db.get_user_by_id.return_value = None # user deleted
|
||||
|
||||
response = client.get("/auth/me", headers={"X-API-Key": plaintext})
|
||||
|
||||
assert response.status_code == 401
|
||||
|
||||
def test_no_auth_at_all_returns_401(self, client, mock_db):
|
||||
"""No auth header at all should return 401."""
|
||||
response = client.get("/auth/me")
|
||||
assert response.status_code == 401
|
||||
|
||||
|
||||
class TestApiKeyFullFlow:
|
||||
"""End-to-end flow: create key, use it, revoke it, try again."""
|
||||
|
||||
def test_create_use_revoke_flow(self, client, mock_db):
|
||||
"""Simulate full lifecycle of an API key."""
|
||||
user = _make_user()
|
||||
mock_db.get_user_by_id.return_value = user
|
||||
|
||||
# Step 1: Create key
|
||||
mock_db.create_api_key.return_value = {
|
||||
"id": 10,
|
||||
"user_id": user["id"],
|
||||
"label": "test",
|
||||
"created_at": datetime(2025, 6, 1, tzinfo=timezone.utc),
|
||||
}
|
||||
|
||||
create_resp = client.post(
|
||||
"/auth/apikeys",
|
||||
json={"label": "test"},
|
||||
headers=_auth_header(user),
|
||||
)
|
||||
assert create_resp.status_code == 200
|
||||
plaintext = create_resp.json()["key"]
|
||||
|
||||
# Capture the hash that was stored
|
||||
call_args = mock_db.create_api_key.call_args
|
||||
stored_hash = call_args.kwargs.get("key_hash") or call_args[0][1]
|
||||
|
||||
# Step 2: Use key on protected endpoint
|
||||
mock_db.get_all_api_key_hashes.return_value = [
|
||||
{"key_hash": stored_hash, "user_id": user["id"]},
|
||||
]
|
||||
|
||||
use_resp = client.get("/auth/me", headers={"X-API-Key": plaintext})
|
||||
assert use_resp.status_code == 200
|
||||
assert use_resp.json()["email"] == user["email"]
|
||||
|
||||
# Step 3: Revoke key
|
||||
mock_db.delete_api_key.return_value = True
|
||||
revoke_resp = client.delete("/auth/apikeys/10", headers=_auth_header(user))
|
||||
assert revoke_resp.status_code == 200
|
||||
|
||||
# Step 4: Try using revoked key
|
||||
mock_db.get_all_api_key_hashes.return_value = [] # key removed from DB
|
||||
rejected_resp = client.get("/auth/me", headers={"X-API-Key": plaintext})
|
||||
assert rejected_resp.status_code == 401
|
||||
|
||||
|
||||
class TestApiKeyHelpers:
|
||||
"""Unit tests for key generation and hashing helpers."""
|
||||
|
||||
def test_generate_api_key_length(self):
|
||||
"""Generated key should be 64 hex characters (32 bytes)."""
|
||||
key = generate_api_key()
|
||||
assert len(key) == 64
|
||||
# Should be valid hex
|
||||
int(key, 16)
|
||||
|
||||
def test_generate_api_key_uniqueness(self):
|
||||
"""Two generated keys should be different."""
|
||||
k1 = generate_api_key()
|
||||
k2 = generate_api_key()
|
||||
assert k1 != k2
|
||||
|
||||
def test_hash_and_verify(self):
|
||||
"""hash_api_key and verify_api_key should round-trip correctly."""
|
||||
key = generate_api_key()
|
||||
hashed = hash_api_key(key)
|
||||
assert verify_api_key(key, hashed)
|
||||
assert not verify_api_key("wrong-key", hashed)
|
||||
Reference in New Issue
Block a user