API Tokens#

API token generation, storage, verification, and usage tracking for service authentication.

Token Generation#

import secrets
import hashlib

def generate_api_token() -> tuple[str, str]:
    # Generate secure random token
    raw_token = secrets.token_urlsafe(32)
    full_token = f"ops_api_token_{raw_token}"

    # Hash for storage
    token_hash = hashlib.sha256(full_token.encode()).hexdigest()

    return full_token, token_hash

Important: Raw token shown once, only hash stored in database.

Token Storage#

Database schema:

CREATE TABLE api_token (
    id SERIAL PRIMARY KEY,
    user_id INTEGER REFERENCES "user"(id),
    name VARCHAR(255),
    token_hash VARCHAR(64) UNIQUE,
    prefix VARCHAR(16),
    last_used TIMESTAMP,
    usage_count INTEGER DEFAULT 0,
    is_active BOOLEAN DEFAULT true,
    expires_at TIMESTAMP,
    created_at TIMESTAMP DEFAULT NOW()
);

Token Verification#

Token verification now returns both user and token object, and enforces scopes:

def verify_api_token(
    token: str,
    db: Session,
    request: Request = None,
    required_scopes: Optional[List[str]] = None
) -> Optional[tuple[User, ApiToken]]:
    # Hash provided token
    token_hash = hashlib.sha256(token.encode()).hexdigest()

    # Lookup in database
    api_token = db.query(ApiToken).filter(
        ApiToken.token_hash == token_hash,
        ApiToken.active == True
    ).first()

    if not api_token or not api_token.is_valid():
        return None

    # Enforce token scopes if required
    if required_scopes:
        token_scopes = set(api_token.scopes or [])
        # Check if token has required scopes (with wildcard support)
        # ...

    # Update usage tracking
    api_token.last_used_at = datetime.now(timezone.utc)
    api_token.usage_count += 1
    if request:
        api_token.last_used_ip = request.client.host
    db.commit()

    return (api_token.user, api_token)

Token Management#

Get available scopes:

curl http://localhost:8000/api/tokens/scopes \
     -H "Authorization: Bearer YOUR_JWT"

Create token (token shown only once):

curl -X POST http://localhost:8000/api/tokens/ \
     -H "Authorization: Bearer YOUR_JWT" \
     -H "Content-Type: application/json" \
     -d '{
       "name": "Observatory Script",
       "scopes": ["read:observations", "write:data"],
       "expires_in_days": 365
     }'

List tokens:

curl http://localhost:8000/api/tokens/ \
     -H "Authorization: Bearer YOUR_JWT"

Get token details:

curl http://localhost:8000/api/tokens/123 \
     -H "Authorization: Bearer YOUR_JWT"

Update token:

curl -X PUT http://localhost:8000/api/tokens/123 \
     -H "Authorization: Bearer YOUR_JWT" \
     -H "Content-Type: application/json" \
     -d '{
       "name": "Updated Name",
       "scopes": ["read:observations"],
       "expires_in_days": 180
     }'

Get token usage statistics:

curl http://localhost:8000/api/tokens/123/usage \
     -H "Authorization: Bearer YOUR_JWT"

Regenerate token (revokes old, creates new):

curl -X POST http://localhost:8000/api/tokens/123/regenerate \
     -H "Authorization: Bearer YOUR_JWT"

Revoke token:

curl -X DELETE http://localhost:8000/api/tokens/123 \
     -H "Authorization: Bearer YOUR_JWT"

Bulk revoke tokens:

curl -X POST http://localhost:8000/api/tokens/bulk-revoke \
     -H "Authorization: Bearer YOUR_JWT" \
     -H "Content-Type: application/json" \
     -d '{"token_ids": [1, 2, 3]}'

Export tokens (for backup/audit):

curl http://localhost:8000/api/tokens/export \
     -H "Authorization: Bearer YOUR_JWT"

Token Scopes#

API tokens support fine-grained permission scopes:

  • read:observations - Read observation data

  • write:observations - Create/update observations

  • read:data - Read data files

  • write:data - Create/update data files

  • read:instruments - Read instrument configurations

  • read:sources - Read source catalog

  • read:programs - Read observing programs

Scopes are enforced during token verification. Wildcard scopes (e.g., read:*) match all specific scopes.

Service Account Tokens#

Certain endpoints require service account tokens (not JWT tokens):

  • POST /executed_obs_units/start

  • PUT /executed_obs_units/{id}/finish

  • POST /raw_data_files/

  • POST /raw_data_files/bulk

  • PUT /raw_data_files/{id}

These endpoints use get_service_user() dependency which: * Rejects JWT tokens * Only accepts API tokens * Requires user to have “service” role

Usage in Scripts#

import requests

API_TOKEN = "ops_api_token_..."  # Service account token
BASE_URL = "http://api.example.com"

headers = {"Authorization": f"Bearer {API_TOKEN}"}

# Service endpoints require service account tokens
response = requests.post(
    f"{BASE_URL}/executed_obs_units/start",
    headers=headers,
    json=observation_data
)

Next Steps#