gistfile1.txt

# cfg file in yaml (loaded at startup)
# users:
#  jbalcas:
#    name: Justas Balcas
#    api_key: a123456789b
#    client_ip: 1.2.3.4
#  xiyang:
#    name: Xi Yang
#    api_key: b987654321a
#    client_ip: 1.2.3.4

class FakeUserDatabase:
    """In-memory user database. Used for development and testing purposes only."""

    def __init__(self):
        self.users = {}
        cfg = getUserConfig() # Custom function to load yaml
        for user, userdict in cfg.get("users", {}).items():
            self.users[user] = account_models.User(
                id=user,
                name=userdict["name"],
                api_key=userdict["api_key"],
                client_ip=userdict["client_ip"],
            )

    def validate(self, token: str) -> dict:
        """Validate token and return claims"""
        for user in self.users.values():
            if user.api_key == token:
                return {
                    "sub": user.id,
                    "preferred_username": user.id,
                    "name": user.name,
                }
        raise HTTPException(status_code=403, detail="Invalid token")

class UserDatabase:
    """Keycloak-backed user database"""

    def __init__(self):
        self.validator = FakeUserDatabase()

    async def get_current_user(self, api_key: str, client_ip: str | None) -> str:
        """Return current user ID based on API key"""
        token = extract_api_key(api_key)
        claims = self.validator.validate(token)
        return claims["preferred_username"]

    async def get_user(
        self, user_id: str, api_key: str, client_ip: str | None
    ) -> account_models.User:
        """Return user object for given user ID and API key"""
        token = extract_api_key(api_key)
        claims = self.validator.validate(token)

        if claims["preferred_username"] != user_id:
            raise HTTPException(status_code=403, detail="User mismatch")

        return account_models.User(
            id=claims["preferred_username"],
            name=claims["name"],
            api_key="",
            client_ip=client_ip,
        )

class AuthMixin(AuthenticatedAdapter, UserDatabase):
    """Mixin class to provide authentication methods required by IRI adapters"""

    get_current_user = UserDatabase.get_current_user
    get_user = UserDatabase.get_user