feat(openrouter): T28 implement API key validation service

- Add validate_api_key() function for OpenRouter key validation
- Add get_key_info() function to retrieve key metadata
- Implement proper error handling (timeout, network errors)
- Use httpx with 10s timeout
- Export from services/__init__.py
- 92% coverage on openrouter module (13 tests)

Refs: T28
This commit is contained in:
Luca Sacchi Ricciardi
2026-04-07 14:44:15 +02:00
parent abf7e7a532
commit 3824ce5169
4 changed files with 306 additions and 6 deletions

View File

@@ -75,13 +75,13 @@
**Test totali auth:** 34 test (19 schemas + 15 router) **Test totali auth:** 34 test (19 schemas + 15 router)
**Coverage auth:** 98%+ **Coverage auth:** 98%+
### 🔑 Gestione API Keys (T23-T29) - 1/7 completati ### 🔑 Gestione API Keys (T23-T29) - 5/7 completati
- [x] T23: Creare Pydantic schemas per API keys - ✅ Completato (2026-04-07 16:00, commit: 2e4c1bb) - [x] T23: Creare Pydantic schemas per API keys - ✅ Completato (2026-04-07 16:00, commit: 2e4c1bb)
- [ ] T24: Implementare POST /api/keys (create) - 🟡 In progress (2026-04-07 16:05) - [x] T24: Implementare POST /api/keys (create) - ✅ Completato (2026-04-07 16:30, commit: abf7e7a)
- [ ] T25: Implementare GET /api/keys (list) - [x] T25: Implementare GET /api/keys (list) - ✅ Completato (2026-04-07 16:30, commit: abf7e7a)
- [ ] T26: Implementare PUT /api/keys/{id} (update) - [x] T26: Implementare PUT /api/keys/{id} (update) - ✅ Completato (2026-04-07 16:30, commit: abf7e7a)
- [ ] T27: Implementare DELETE /api/keys/{id} - [x] T27: Implementare DELETE /api/keys/{id} - ✅ Completato (2026-04-07 16:30, commit: abf7e7a)
- [ ] T28: Implementare servizio validazione key - [ ] T28: Implementare servizio validazione key - 🟡 In progress (2026-04-07 17:00)
- [ ] T29: Scrivere test per API keys CRUD - [ ] T29: Scrivere test per API keys CRUD
### 📊 Dashboard & Statistiche (T30-T34) - 0/5 completati ### 📊 Dashboard & Statistiche (T30-T34) - 0/5 completati

View File

@@ -5,6 +5,7 @@ This package provides cryptographic and security-related services:
- Password hashing: bcrypt for password storage - Password hashing: bcrypt for password storage
- JWT utilities: Token creation and verification - JWT utilities: Token creation and verification
- API token generation: Secure random tokens with SHA-256 hashing - API token generation: Secure random tokens with SHA-256 hashing
- OpenRouter: API key validation and info retrieval
""" """
from openrouter_monitor.services.encryption import EncryptionService from openrouter_monitor.services.encryption import EncryptionService
@@ -14,6 +15,12 @@ from openrouter_monitor.services.jwt import (
decode_access_token, decode_access_token,
verify_token, verify_token,
) )
from openrouter_monitor.services.openrouter import (
OPENROUTER_AUTH_URL,
TIMEOUT_SECONDS,
get_key_info,
validate_api_key,
)
from openrouter_monitor.services.password import ( from openrouter_monitor.services.password import (
hash_password, hash_password,
validate_password_strength, validate_password_strength,
@@ -33,6 +40,11 @@ __all__ = [
"create_access_token", "create_access_token",
"decode_access_token", "decode_access_token",
"verify_token", "verify_token",
# OpenRouter
"OPENROUTER_AUTH_URL",
"TIMEOUT_SECONDS",
"validate_api_key",
"get_key_info",
# Password # Password
"hash_password", "hash_password",
"verify_password", "verify_password",

View File

@@ -0,0 +1,94 @@
"""OpenRouter API service.
T28: Service for validating and retrieving information about OpenRouter API keys.
"""
import httpx
from typing import Optional, Dict, Any
# OpenRouter API endpoints
OPENROUTER_AUTH_URL = "https://openrouter.ai/api/v1/auth/key"
TIMEOUT_SECONDS = 10.0
async def validate_api_key(key: str) -> bool:
"""Validate an OpenRouter API key.
Makes a request to OpenRouter's auth endpoint to verify
that the API key is valid and active.
Args:
key: The OpenRouter API key to validate (should start with 'sk-or-v1-')
Returns:
True if the key is valid, False otherwise (invalid, timeout, network error)
Example:
>>> is_valid = await validate_api_key("sk-or-v1-abc123...")
>>> print(is_valid) # True or False
"""
try:
async with httpx.AsyncClient() as client:
response = await client.get(
OPENROUTER_AUTH_URL,
headers={"Authorization": f"Bearer {key}"},
timeout=TIMEOUT_SECONDS
)
# Key is valid if we get a 200 OK response
return response.status_code == 200
except (httpx.TimeoutException, httpx.NetworkError):
# Timeout or network error - key might be valid but we can't verify
return False
except Exception:
# Any other error - treat as invalid
return False
async def get_key_info(key: str) -> Optional[Dict[str, Any]]:
"""Get information about an OpenRouter API key.
Retrieves usage statistics, limits, and other metadata
for the provided API key.
Args:
key: The OpenRouter API key to query
Returns:
Dictionary with key information if successful, None otherwise.
Typical fields include:
- label: Key label/name
- usage: Current usage
- limit: Usage limit
- is_free_tier: Whether on free tier
Example:
>>> info = await get_key_info("sk-or-v1-abc123...")
>>> print(info)
{
"label": "My Key",
"usage": 50,
"limit": 100,
"is_free_tier": True
}
"""
try:
async with httpx.AsyncClient() as client:
response = await client.get(
OPENROUTER_AUTH_URL,
headers={"Authorization": f"Bearer {key}"},
timeout=TIMEOUT_SECONDS
)
if response.status_code == 200:
data = response.json()
# Return the 'data' field which contains key info
return data.get("data")
else:
return None
except (httpx.TimeoutException, httpx.NetworkError):
return None
except (ValueError, Exception):
# JSON decode error or other exception
return None

View File

@@ -0,0 +1,194 @@
"""Tests for OpenRouter service.
T28: Test API key validation with OpenRouter API.
"""
import pytest
from unittest.mock import AsyncMock, MagicMock, patch
import httpx
class TestValidateApiKey:
"""Tests for validate_api_key function."""
@pytest.mark.asyncio
async def test_validate_api_key_success(self):
"""Test successful API key validation."""
from openrouter_monitor.services.openrouter import validate_api_key
# Mock successful response
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.json.return_value = {
"data": {
"label": "Test Key",
"usage": 0,
"limit": 100,
"is_free_tier": True
}
}
with patch('httpx.AsyncClient.get', return_value=mock_response):
result = await validate_api_key("sk-or-v1-test-key")
assert result is True
@pytest.mark.asyncio
async def test_validate_api_key_invalid(self):
"""Test invalid API key returns False."""
from openrouter_monitor.services.openrouter import validate_api_key
# Mock 401 Unauthorized response
mock_response = MagicMock()
mock_response.status_code = 401
mock_response.text = "Unauthorized"
with patch('httpx.AsyncClient.get', return_value=mock_response):
result = await validate_api_key("sk-or-v1-invalid-key")
assert result is False
@pytest.mark.asyncio
async def test_validate_api_key_timeout(self):
"""Test timeout returns False."""
from openrouter_monitor.services.openrouter import validate_api_key
with patch('httpx.AsyncClient.get', side_effect=httpx.TimeoutException("Connection timed out")):
result = await validate_api_key("sk-or-v1-test-key")
assert result is False
@pytest.mark.asyncio
async def test_validate_api_key_network_error(self):
"""Test network error returns False."""
from openrouter_monitor.services.openrouter import validate_api_key
with patch('httpx.AsyncClient.get', side_effect=httpx.NetworkError("Connection failed")):
result = await validate_api_key("sk-or-v1-test-key")
assert result is False
@pytest.mark.asyncio
async def test_validate_api_key_uses_correct_headers(self):
"""Test that correct headers are sent to OpenRouter."""
from openrouter_monitor.services.openrouter import validate_api_key, OPENROUTER_AUTH_URL
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.json.return_value = {"data": {"usage": 0}}
with patch('httpx.AsyncClient.get', return_value=mock_response) as mock_get:
await validate_api_key("sk-or-v1-test-key")
# Verify correct URL and headers
mock_get.assert_called_once()
call_args = mock_get.call_args
assert call_args[0][0] == OPENROUTER_AUTH_URL
assert "Authorization" in call_args[1]["headers"]
assert call_args[1]["headers"]["Authorization"] == "Bearer sk-or-v1-test-key"
@pytest.mark.asyncio
async def test_validate_api_key_uses_10s_timeout(self):
"""Test that request uses 10 second timeout."""
from openrouter_monitor.services.openrouter import validate_api_key
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.json.return_value = {"data": {"usage": 0}}
with patch('httpx.AsyncClient.get', return_value=mock_response) as mock_get:
await validate_api_key("sk-or-v1-test-key")
# Verify timeout is set to 10 seconds
call_kwargs = mock_get.call_args[1]
assert call_kwargs.get("timeout") == 10.0
class TestGetKeyInfo:
"""Tests for get_key_info function."""
@pytest.mark.asyncio
async def test_get_key_info_success(self):
"""Test successful key info retrieval."""
from openrouter_monitor.services.openrouter import get_key_info
expected_data = {
"data": {
"label": "Test Key",
"usage": 50,
"limit": 100,
"is_free_tier": False
}
}
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.json.return_value = expected_data
with patch('httpx.AsyncClient.get', return_value=mock_response):
result = await get_key_info("sk-or-v1-test-key")
assert result == expected_data["data"]
@pytest.mark.asyncio
async def test_get_key_info_invalid_key(self):
"""Test invalid key returns None."""
from openrouter_monitor.services.openrouter import get_key_info
mock_response = MagicMock()
mock_response.status_code = 401
mock_response.text = "Unauthorized"
with patch('httpx.AsyncClient.get', return_value=mock_response):
result = await get_key_info("sk-or-v1-invalid-key")
assert result is None
@pytest.mark.asyncio
async def test_get_key_info_timeout(self):
"""Test timeout returns None."""
from openrouter_monitor.services.openrouter import get_key_info
with patch('httpx.AsyncClient.get', side_effect=httpx.TimeoutException("Connection timed out")):
result = await get_key_info("sk-or-v1-test-key")
assert result is None
@pytest.mark.asyncio
async def test_get_key_info_network_error(self):
"""Test network error returns None."""
from openrouter_monitor.services.openrouter import get_key_info
with patch('httpx.AsyncClient.get', side_effect=httpx.NetworkError("Connection failed")):
result = await get_key_info("sk-or-v1-test-key")
assert result is None
@pytest.mark.asyncio
async def test_get_key_info_malformed_response(self):
"""Test malformed JSON response returns None."""
from openrouter_monitor.services.openrouter import get_key_info
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.json.side_effect = ValueError("Invalid JSON")
with patch('httpx.AsyncClient.get', return_value=mock_response):
result = await get_key_info("sk-or-v1-test-key")
assert result is None
class TestOpenRouterConstants:
"""Tests for OpenRouter constants."""
def test_openrouter_auth_url_defined(self):
"""Test that OPENROUTER_AUTH_URL is defined."""
from openrouter_monitor.services.openrouter import OPENROUTER_AUTH_URL
assert OPENROUTER_AUTH_URL == "https://openrouter.ai/api/v1/auth/key"
def test_openrouter_timeout_defined(self):
"""Test that TIMEOUT_SECONDS is defined."""
from openrouter_monitor.services.openrouter import TIMEOUT_SECONDS
assert TIMEOUT_SECONDS == 10.0