| 12 | |
| 13 | |
| 14 | class AuthInterceptor(grpc.ServerInterceptor): |
| 15 | def intercept_service(self, continuation, handler_call_details): |
| 16 | sm = get_security_manager() |
| 17 | |
| 18 | if sm is not None: |
| 19 | auth_manager = get_auth_manager() |
| 20 | access_token = auth_manager.token_extractor.extract_access_token( |
| 21 | metadata=dict(handler_call_details.invocation_metadata) |
| 22 | ) |
| 23 | |
| 24 | logger.debug( |
| 25 | f"Fetching user details for token of length: {len(access_token)}" |
| 26 | ) |
| 27 | current_user = asyncio.run( |
| 28 | auth_manager.token_parser.user_details_from_access_token(access_token) |
| 29 | ) |
| 30 | logger.debug(f"User is: {current_user}") |
| 31 | sm.set_current_user(current_user) |
| 32 | |
| 33 | return continuation(handler_call_details) |
no outgoing calls
no test coverage detected