Coverage for /usr/local/lib/python3.10/site-packages/opal_common-0.0.0-py3.10.egg/opal_common/authentication/authz.py: 52%

1from opal_common.authentication.deps import JWTAuthenticator 

2from opal_common.authentication.types import JWTClaims 

3from opal_common.authentication.verifier import Unauthorized 

4from opal_common.schemas.data import DataUpdate 

5from opal_common.schemas.security import PeerType 

6 

7 

8def require_peer_type( 

9 authenticator: JWTAuthenticator, claims: JWTClaims, required_type: PeerType 

10): 

11 if not authenticator.enabled: 11 ↛ 12line 11 didn't jump to line 12 because the condition on line 11 was never true

12 return 

13 

14 peer_type = claims.get("peer_type", None) 

15 if peer_type is None: 15 ↛ 16line 15 didn't jump to line 16 because the condition on line 15 was never true

16 raise Unauthorized(description="Missing 'peer_type' claim for OPAL jwt token") 

17 try: 

18 type = PeerType(peer_type) 

19 except ValueError: 

20 raise Unauthorized( 

21 description=f"Invalid 'peer_type' claim for OPAL jwt token: {peer_type}" 

22 ) 

23 

24 if type != required_type: 24 ↛ 25line 24 didn't jump to line 25 because the condition on line 24 was never true

25 raise Unauthorized( 

26 description=f"Incorrect 'peer_type' claim for OPAL jwt token: {str(type)}, expected: {str(required_type)}" 

27 ) 

28 

29 

30def restrict_optional_topics_to_publish( 

31 authenticator: JWTAuthenticator, claims: JWTClaims, update: DataUpdate 

32): 

33 if not authenticator.enabled: 33 ↛ 34line 33 didn't jump to line 34 because the condition on line 33 was never true

34 return 

35 

36 if "permitted_topics" not in claims: 36 ↛ 39line 36 didn't jump to line 39 because the condition on line 36 was always true

37 return 

38 

39 for entry in update.entries: 

40 unauthorized_topics = set(entry.topics).difference(claims["permitted_topics"]) 

41 if unauthorized_topics: 

42 raise Unauthorized( 

43 description=f"Invalid 'topics' to publish {unauthorized_topics}" 

44 )