"""Handles user authentication to obtain access to tango gql services."""
import logging
from typing import Any, NamedTuple
import requests
from .configuration import Environment
logger = logging.getLogger(__name__)
[docs]class AuthenticatedUser(NamedTuple):
"""Bundle authenticated user values as a singel object."""
cookies: Any
username: str
password: str
@property
def auth(self) -> bool:
"""Whether the object represents a successful authentication.
:return: True if authentication successful
"""
return self.cookies != ""
[docs]class AuthException(Exception):
"""Raised when authentication unsuccessful."""
pass
[docs]class Authenticator:
"""Generates an authenticated user."""
def __init__(self, env: Environment) -> None:
"""Initialise object.
:param env: Host environment values from which to create authentication
:type env: Environment
"""
self._environment = env
self._credentials = self._environment.get_credentials()
[docs] def get_authenticated_user(self) -> AuthenticatedUser:
"""Return authenticated user values for gaining access to tango gql services.
:raises AuthException: when authentication unsuccessful
:return: authenticated user values as a AuthenticatedUser object.
"""
if self._credentials:
if self._environment.tango_bridge_ip:
host = self._environment.tango_bridge_ip
else:
host = self._environment.kubehost
url = f"http://{host}/auth/login"
try:
result = requests.post(url, json=self._credentials.asdict())
except requests.exceptions.ConnectionError as exception:
raise AuthException(
f"Unable to authenticate on {url} as connection is not available"
) from exception
if result.status_code != 200:
raise AuthException(
f"Unable to authenticate on {url} message_code: "
f"{result.status_code}"
)
if cookie := result.cookies.get_dict().get("taranta_jwt"):
cookies = {"taranta_jwt": cookie}
else:
webjive_cookie = result.cookies.get_dict()["webjive_jwt"]
cookies = {"webjive_jwt": webjive_cookie}
logger.info("user authenticated")
return AuthenticatedUser(
cookies, self._credentials.username, self._credentials.password
)
return AuthenticatedUser("", "", "")