Coverage for /usr/local/lib/python3.10/site-packages/opal_server-0.0.0-py3.10.egg/opal_server/security/api.py: 89%

1from datetime import datetime 

2 

3from fastapi import APIRouter, Depends, HTTPException, status 

4from opal_common.authentication.deps import StaticBearerAuthenticator 

5from opal_common.authentication.signer import JWTSigner 

6from opal_common.logger import logger 

7from opal_common.schemas.security import AccessToken, AccessTokenRequest, TokenDetails 

8 

9 

10def init_security_router(signer: JWTSigner, authenticator: StaticBearerAuthenticator): 

11 router = APIRouter() 

12 

13 @router.post( 

14 "/token", 

15 status_code=status.HTTP_200_OK, 

16 response_model=AccessToken, 

17 dependencies=[Depends(authenticator)], 

18 ) 

19 async def generate_new_access_token(req: AccessTokenRequest): 

20 if not signer.enabled: 20 ↛ 21line 20 didn't jump to line 21 because the condition on line 20 was never true

21 raise HTTPException( 

22 status_code=status.HTTP_503_SERVICE_UNAVAILABLE, 

23 detail="opal server was not configured with security, cannot generate tokens!", 

24 ) 

25 

26 claims = {"peer_type": req.type.value, **req.claims} 

27 token = signer.sign(sub=req.id, token_lifetime=req.ttl, custom_claims=claims) 

28 logger.info(f"Generated opal token: peer_type={req.type.value}") 

29 return AccessToken( 

30 token=token, 

31 details=TokenDetails( 

32 id=req.id, 

33 type=req.type, 

34 expired=datetime.utcnow() + req.ttl, 

35 claims=claims, 

36 ), 

37 ) 

38 

39 return router