forked from sr2/cloud-api
luxferre
c689ac1e10
All changes are either: - Correcting tabs - Adding/removing line breaks - Adding trailing commas
35 lines
906 B
Python
35 lines
906 B
Python
"""
|
|
Business logic reusable functions related to IAM
|
|
|
|
Exports:
|
|
- service_key_dependency: bool: verifies request headers contain the correct api key for the service
|
|
"""
|
|
|
|
from typing import Annotated
|
|
|
|
from src.service.models import Service
|
|
from src.database import db_dependency
|
|
from src.schemas import ResourceName
|
|
from src.auth.exceptions import UnauthorizedException
|
|
|
|
from fastapi import Request, Depends
|
|
|
|
|
|
def valid_service_key(db: db_dependency, request: Request, rn: ResourceName) -> bool:
|
|
api_key = request.headers.get("X-API-Key", None)
|
|
if not api_key:
|
|
raise UnauthorizedException("Missing API key")
|
|
service = rn.service
|
|
result = (
|
|
db.query(Service)
|
|
.filter(Service.name == service)
|
|
.filter(Service.api_key == api_key)
|
|
.first()
|
|
)
|
|
if result is None:
|
|
raise UnauthorizedException("Invalid API key")
|
|
|
|
return True
|
|
|
|
|
|
service_key_dependency = Annotated[bool, Depends(valid_service_key)]
|