# Path: virtool/http/client.py
# class UserClient(AbstractClient):
#     def __init__(
#         self,
#         db,
#         administrator: bool,
#         force_reset: bool,
#         groups: Sequence[str],
#         permissions: Dict[str, bool],
#         user_id: Union[str, None],
#         authenticated: bool,
#         session_id: Optional[str] = None,
#     ):
#         self._db = db
#         self._force_reset = force_reset
#         self._administrator = administrator
#         self._authenticated = authenticated
#         self.groups = groups
#         self.permissions = permissions
#         self.user_id = user_id
#         self.session_id = session_id
# 
#     @property
#     def authenticated(self) -> bool:
#         return self._authenticated
# 
#     @property
#     def administrator(self) -> bool:
#         return self._administrator
# 
#     @property
#     def force_reset(self) -> bool:
#         return self._force_reset
# 
#     def has_permission(self, permission: str) -> bool:
#         return self.permissions.get(permission, False)
# 
#     async def has_right_on_analysis(self, analysis_id: str, right: Right) -> bool:
#         return True
# 
#     async def has_right_on_hmms(self, right: Right):
#         if right == READ:
#             return True
# 
#         if right == MODIFY or right == REMOVE:
#             return self.has_permission("modify_hmm")
# 
#     async def has_right_on_index(self, index_id: str, right: Right) -> bool:
#         return True if right == READ else False
# 
#     async def has_right_on_sample(self, sample_id: str, right: Right) -> bool:
#         if self.administrator:
#             return True
# 
#         sample = await self._db.find_one(sample_id, {"quality": False})
# 
#         if self.user_id == sample["user"]["id"]:
#             return True
# 
#         is_group_member = sample["group"] and sample["group"] in self.groups
# 
#         if right == READ:
#             return sample["all_read"] or (is_group_member and sample["group_read"])
# 
#         if right == MODIFY or right == REMOVE:
#             return sample["all_write"] or (is_group_member and sample["group_write"])
# 
#     async def has_right_on_reference(self, reference_id: str, right: Right):
#         return False
# 
#     async def has_right_on_subtraction(self, subtraction_id: str, right: Right):
#         """
#         Check whether the authenticated user has the passed ``right`` on a subtraction.
# 
#         User rights on subtractions are based on group permissions.
# 
#         """
#         if right == READ:
#             return True
# 
#         if right == MODIFY:
#             return self.has_permission("modify_subtraction")
# 
#         if right == REMOVE:
#             return self.has_permission("delete_subtraction")
# 
#     async def has_right_on_upload(self, upload_id: str, right: Right) -> bool:
#         if right == REMOVE:
#             return self.has_permission("remove_file")
# 
#         if right == MODIFY:
#             return False
# 
#         return True
# 
# Path: virtool/oidc/utils.py
# async def update_jwk_args(app: Application, token: jwt) -> JWKArgs:
#     """
#     Update jwk_args and store in environment variable
# 
#     :param token: JWT token to validate and decode
#     :param app: the app object
#     :return: jwk_args
#     """
#     header = await app["run_in_thread"](get_unverified_header, token)
#     authority = app["b2c"].authority
#     resp = await app["client"].get(
#         f"{authority}/discovery/v2.0/keys", allow_redirects=True
#     )
#     jwks = json.loads(await resp.text())
# 
#     jwk = await get_matching_jwk(jwks, header["kid"])
# 
#     jwk_args = JWKArgs(
#         kty=jwk["kty"], kid=jwk["kid"], use=jwk["use"], n=jwk["n"], e=jwk["e"]
#     )
# 
#     app["b2c"].jwk_args = jwk_args
# 
#     return jwk_args
# 
# async def validate_token(app: Application, token: jwt) -> dict:
#     """
#     Validate token and return claims using JWK_ARGS environment variable.
# 
#     :param token: JWT token to validate
#     :param app: the application object
#     :return: JWT claims
#     """
# 
#     jwk_args = app["b2c"].jwk_args or await update_jwk_args(app, token)
# 
#     return decode(
#         token,
#         asdict(jwk_args),
#         algorithms=["RS256"],
#         audience=app["config"].b2c_client_id,
#     )


from typing import Callable, Tuple
from aiohttp import BasicAuth, web
from aiohttp.web import HTTPFound, Request, Response
from aiohttp.web_exceptions import HTTPUnauthorized
from jose import ExpiredSignatureError
from jose.exceptions import JWTClaimsError, JWTError
from virtool.http.client import UserClient
from virtool.http.utils import set_session_id_cookie
from virtool.oidc.utils import update_jwk_args, validate_token
from virtool.users.db import B2CUserAttributes, find_or_create_b2c_user
from virtool.users.sessions import clear_reset_code, create_session, get_session
from virtool.utils import random_alphanumeric
import virtool.errors
import virtool.utils
            administrator=False,
            force_reset=False,
            groups=list(),
            permissions=dict(),
            user_id=None,
            authenticated=False,
        )

        return await handler(req)

    if req.headers.get("AUTHORIZATION"):
        return await authenticate_with_key(req, handler)

    if req.app["config"].use_b2c:
        return await authenticate_with_b2c(req, handler)

    # Get session information from cookies.
    session_id = req.cookies.get("session_id")
    session_token = req.cookies.get("session_token")

    session, session_token = await get_session(db, session_id, session_token)

    ip = get_ip(req)

    if session is None:
        session, session_token = await create_session(db, ip)

    session_id = session["_id"]

    if session_token:

# Path: virtool/http/client.py
# class UserClient(AbstractClient):
#     def __init__(
#         self,
#         db,
#         administrator: bool,
#         force_reset: bool,
#         groups: Sequence[str],
#         permissions: Dict[str, bool],
#         user_id: Union[str, None],
#         authenticated: bool,
#         session_id: Optional[str] = None,
#     ):
#         self._db = db
#         self._force_reset = force_reset
#         self._administrator = administrator
#         self._authenticated = authenticated
#         self.groups = groups
#         self.permissions = permissions
#         self.user_id = user_id
#         self.session_id = session_id
# 
#     @property
#     def authenticated(self) -> bool:
#         return self._authenticated
# 
#     @property
#     def administrator(self) -> bool:
#         return self._administrator
# 
#     @property
#     def force_reset(self) -> bool:
#         return self._force_reset
# 
#     def has_permission(self, permission: str) -> bool:
#         return self.permissions.get(permission, False)
# 
#     async def has_right_on_analysis(self, analysis_id: str, right: Right) -> bool:
#         return True
# 
#     async def has_right_on_hmms(self, right: Right):
#         if right == READ:
#             return True
# 
#         if right == MODIFY or right == REMOVE:
#             return self.has_permission("modify_hmm")
# 
#     async def has_right_on_index(self, index_id: str, right: Right) -> bool:
#         return True if right == READ else False
# 
#     async def has_right_on_sample(self, sample_id: str, right: Right) -> bool:
#         if self.administrator:
#             return True
# 
#         sample = await self._db.find_one(sample_id, {"quality": False})
# 
#         if self.user_id == sample["user"]["id"]:
#             return True
# 
#         is_group_member = sample["group"] and sample["group"] in self.groups
# 
#         if right == READ:
#             return sample["all_read"] or (is_group_member and sample["group_read"])
# 
#         if right == MODIFY or right == REMOVE:
#             return sample["all_write"] or (is_group_member and sample["group_write"])
# 
#     async def has_right_on_reference(self, reference_id: str, right: Right):
#         return False
# 
#     async def has_right_on_subtraction(self, subtraction_id: str, right: Right):
#         """
#         Check whether the authenticated user has the passed ``right`` on a subtraction.
# 
#         User rights on subtractions are based on group permissions.
# 
#         """
#         if right == READ:
#             return True
# 
#         if right == MODIFY:
#             return self.has_permission("modify_subtraction")
# 
#         if right == REMOVE:
#             return self.has_permission("delete_subtraction")
# 
#     async def has_right_on_upload(self, upload_id: str, right: Right) -> bool:
#         if right == REMOVE:
#             return self.has_permission("remove_file")
# 
#         if right == MODIFY:
#             return False
# 
#         return True
# 
# Path: virtool/oidc/utils.py
# async def update_jwk_args(app: Application, token: jwt) -> JWKArgs:
#     """
#     Update jwk_args and store in environment variable
# 
#     :param token: JWT token to validate and decode
#     :param app: the app object
#     :return: jwk_args
#     """
#     header = await app["run_in_thread"](get_unverified_header, token)
#     authority = app["b2c"].authority
#     resp = await app["client"].get(
#         f"{authority}/discovery/v2.0/keys", allow_redirects=True
#     )
#     jwks = json.loads(await resp.text())
# 
#     jwk = await get_matching_jwk(jwks, header["kid"])
# 
#     jwk_args = JWKArgs(
#         kty=jwk["kty"], kid=jwk["kid"], use=jwk["use"], n=jwk["n"], e=jwk["e"]
#     )
# 
#     app["b2c"].jwk_args = jwk_args
# 
#     return jwk_args
# 
# async def validate_token(app: Application, token: jwt) -> dict:
#     """
#     Validate token and return claims using JWK_ARGS environment variable.
# 
#     :param token: JWT token to validate
#     :param app: the application object
#     :return: JWT claims
#     """
# 
#     jwk_args = app["b2c"].jwk_args or await update_jwk_args(app, token)
# 
#     return decode(
#         token,
#         asdict(jwk_args),
#         algorithms=["RS256"],
#         audience=app["config"].b2c_client_id,
#     )


# Path: virtool/http/auth.py
from typing import Callable, Tuple
from aiohttp import BasicAuth, web
from aiohttp.web import HTTPFound, Request, Response
from aiohttp.web_exceptions import HTTPUnauthorized
from jose import ExpiredSignatureError
from jose.exceptions import JWTClaimsError, JWTError
from virtool.http.client import UserClient
from virtool.http.utils import set_session_id_cookie
from virtool.oidc.utils import update_jwk_args, validate_token
from virtool.users.db import B2CUserAttributes, find_or_create_b2c_user
from virtool.users.sessions import clear_reset_code, create_session, get_session
from virtool.utils import random_alphanumeric
import virtool.errors
import virtool.utils

            administrator=False,
            force_reset=False,
            groups=list(),
            permissions=dict(),
            user_id=None,
            authenticated=False,
        )

        return await handler(req)

    if req.headers.get("AUTHORIZATION"):
        return await authenticate_with_key(req, handler)

    if req.app["config"].use_b2c:
        return await authenticate_with_b2c(req, handler)

    # Get session information from cookies.
    session_id = req.cookies.get("session_id")
    session_token = req.cookies.get("session_token")

    session, session_token = await get_session(db, session_id, session_token)

    ip = get_ip(req)

    if session is None:
        session, session_token = await create_session(db, ip)

    session_id = session["_id"]

    if session_token:

        req["client"] = UserClient(