Middleware that adds Basic Authentication to all requests.
| 29 | |
| 30 | |
| 31 | class BasicAuthMiddleware: |
| 32 | """Middleware that adds Basic Authentication to all requests.""" |
| 33 | |
| 34 | def __init__(self, username: str, password: str) -> None: |
| 35 | self.username = username |
| 36 | self.password = password |
| 37 | self._auth_header = self._encode_credentials() |
| 38 | |
| 39 | def _encode_credentials(self) -> str: |
| 40 | """Encode username and password to base64.""" |
| 41 | credentials = f"{self.username}:{self.password}" |
| 42 | encoded = base64.b64encode(credentials.encode()).decode() |
| 43 | return f"Basic {encoded}" |
| 44 | |
| 45 | async def __call__( |
| 46 | self, |
| 47 | request: ClientRequest, |
| 48 | handler: ClientHandlerType, |
| 49 | ) -> ClientResponse: |
| 50 | """Add Basic Auth header to the request.""" |
| 51 | # Only add auth if not already present |
| 52 | if hdrs.AUTHORIZATION not in request.headers: |
| 53 | request.headers[hdrs.AUTHORIZATION] = self._auth_header |
| 54 | |
| 55 | # Proceed with the request |
| 56 | return await handler(request) |
| 57 | |
| 58 | |
| 59 | class TestServer: |