Source code for api.utils

import jwt

from datetime import timedelta
from onegov.api import ApiApp
from onegov.api.models import ApiException, ApiKey
from onegov.api.token import try_get_encoded_token, jwt_decode
from sedate import utcnow
from webob.exc import HTTPUnauthorized, HTTPClientError

from typing import TYPE_CHECKING
    from onegov.core.request import CoreRequest
    from morepath.request import Response

[docs] def authenticate(request: 'CoreRequest') -> None: try: auth = try_get_encoded_token(request) data = jwt_decode(request, auth) except jwt.ExpiredSignatureError as exception: raise HTTPUnauthorized() from exception except Exception as e: raise ApiException() from e if request.session.query(ApiKey).get(data['id']) is None: raise HTTPClientError()
[docs] def check_rate_limit(request: 'CoreRequest') -> dict[str, str]: """ Checks if the rate limit for the current client. Raises an exception if the rate limit is reached. Returns response headers containing informations about the remaining rate limit. Logged in users don't have rate limits. The same is true for users that have authenticated with a token. """ if request.is_logged_in: return {} if request.authorization: authenticate(request) return {} assert isinstance(, ApiApp) addr = request.client_addr or 'unknown' limit, expiration = requests, timestamp = addr, creator=lambda: (0, utcnow()), ) if (utcnow() - timestamp).seconds < expiration: requests += 1 else: timestamp = utcnow() requests = 1 addr, (requests, timestamp) ) reset = timestamp + timedelta(seconds=expiration) headers = { 'X-RateLimit-Limit': str(limit), 'X-RateLimit-Remaining': str(max(limit - requests, 0)), 'X-RateLimit-Reset': reset.strftime("%a, %d %b %Y %H:%M:%S GMT") } @request.after def add_headers(response: 'Response') -> None: for header in headers.items(): response.headers.add(*header) if requests > limit: headers['Retry-After'] = headers['X-RateLimit-Reset'] raise ApiException( 'Rate limit exceeded', status_code=429, headers=headers ) return headers