📦 核心文件

openhands/app_server/user_auth/user_auth.py — 抽象基类

openhands/app_server/user_auth/default_user_auth.py — OSS 默认实现

enterprise/server/auth/saas_user_auth.py — SaaS Keycloak 实现

enterprise/server/auth/authorization.py — 权限模型

enterprise/server/routes/auth.py — 认证路由

📄 openhands/app_server/user_auth/user_auth.py (第 30-125 行)

class AuthType(Enum):
    COOKIE = 'cookie'
    BEARER = 'bearer'


class UserAuth(ABC):
    """Abstract base class for user authentication.

    This is an extension point in OpenHands that allows applications to provide their own
    authentication mechanisms. Applications can substitute their own implementation by:
    1. Creating a class that inherits from UserAuth
    2. Implementing all required methods
    3. Setting server_config.user_auth_class to the fully qualified name of the class

    The class is instantiated via get_impl() in openhands.app_server.shared.py.
    """

    _settings: Settings | None
    _resolved_settings: Settings | None = None

    @abstractmethod
    async def get_user_id(self) -> str | None:
        """Get the unique identifier for the current user"""

    @abstractmethod
    async def get_user_email(self) -> str | None:
        """Get the email for the current user"""

    @abstractmethod
    async def get_access_token(self) -> SecretStr | None:
        """Get the access token for the current user"""

    @abstractmethod
    async def get_provider_tokens(self) -> PROVIDER_TOKEN_TYPE | None:
        """Get the provider tokens for the current user."""

    @abstractmethod
    async def get_user_settings_store(self) -> SettingsStore:
        """Get the settings store for the current user."""

    @abstractmethod
    async def get_secrets_store(self) -> SecretsStore:
        """Get secrets store"""

    @abstractmethod
    async def get_secrets(self) -> Secrets | None:
        """Get the user's secrets"""

    @abstractmethod
    async def get_mcp_api_key(self) -> str | None:
        """Get an mcp api key for the user"""

    def get_auth_type(self) -> AuthType | None:
        return None

设计要点：

UserAuth 不是一个简单的认证器，它是一个用户上下文的聚合器。它同时管理：

🔹 用户身份（ID、邮箱、访问令牌）

🔹 用户设置（SettingsStore）

🔹 用户密钥（SecretsStore）

🔹 第三方 Provider Token（GitHub、GitLab 等）

📄 openhands/app_server/user_auth/user_auth.py (第 151-168 行)

async def get_user_auth(request: Request) -> UserAuth:
    user_auth: UserAuth | None = getattr(request.state, 'user_auth', None)
    if user_auth:
        return user_auth
    impl_name = server_config.user_auth_class
    impl = get_impl(UserAuth, impl_name)
    user_auth = await impl.get_instance(request)
    if user_auth is None:
        raise ValueError('Failed to get user auth instance')
    request.state.user_auth = user_auth
    return user_auth


async def get_for_user(user_id: str) -> UserAuth:
    impl_name = server_config.user_auth_class
    impl = get_impl(UserAuth, impl_name)
    user_auth = await impl.get_for_user(user_id)
    return user_auth

📄 openhands/app_server/user_auth/default_user_auth.py (第 24-92 行)

@dataclass
class DefaultUserAuth(UserAuth):
    """Default user authentication mechanism"""

    _settings: Settings | None = None
    _settings_store: SettingsStore | None = None
    _secrets_store: SecretsStore | None = None
    _secrets: Secrets | None = None

    async def get_user_id(self) -> str | None:
        """The default implementation does not support multi tenancy,
        so user_id is always None"""
        return None

    async def get_user_email(self) -> str | None:
        """The default implementation does not support multi tenancy,
        so email is always None"""
        return None

    async def get_access_token(self) -> SecretStr | None:
        """The default implementation does not support multi tenancy,
        so access_token is always None"""
        return None

    @classmethod
    async def get_for_user(cls, user_id: str) -> UserAuth:
        assert user_id == 'root'
        return DefaultUserAuth()

简洁即力量：

DefaultUserAuth 只有 92 行代码。它用 assert user_id == 'root' 声明了 OSS 模式只存在一个"超级用户"。Settings 和 Secrets 通过单例 Store 管理，不区分用户。

📄 enterprise/server/auth/saas_user_auth.py (第 57-93 行)

@dataclass
class SaasUserAuth(UserAuth):
    refresh_token: SecretStr
    user_id: str
    email: str | None = None
    email_verified: bool | None = None
    access_token: SecretStr | None = None
    provider_tokens: PROVIDER_TOKEN_TYPE | None = None
    refreshed: bool = False
    settings_store: SaasSettingsStore | None = None
    secrets_store: SaasSecretsStore | None = None
    _settings: Settings | None = None
    _resolved_settings: Settings | None = None
    _secrets: Secrets | None = None
    accepted_tos: bool | None = None
    auth_type: AuthType = AuthType.COOKIE
    # API key context fields - populated when authenticated via API key
    api_key_org_id: UUID | None = None
    api_key_id: int | None = None
    api_key_name: str | None = None
    # Organization context fields - populated lazily via get_org_info()
    _org_id: str | None = None
    _org_name: str | None = None
    _role: str | None = None
    _permissions: list[str] | None = None
    _org_info_loaded: bool = False
    # Per-request X-Org-Id header (raw, unvalidated)
    _x_org_id_header: str | None = None
    # Trusted server-side override used by background resolver contexts
    effective_org_id_override: UUID | None = None
    # Cached result of get_effective_org_id()
    _effective_org_id: UUID | None = None
    _effective_org_id_resolved: bool = False

📄 enterprise/server/routes/auth.py (第 80-131 行)

api_router = APIRouter(prefix='/api')
oauth_router = APIRouter(prefix='/oauth')

token_manager = TokenManager()


def set_response_cookie(
    request: Request,
    response: Response,
    keycloak_access_token: str,
    keycloak_refresh_token: str,
    secure: bool = True,
    accepted_tos: bool = False,
):
    # Create a signed JWT token
    cookie_data = {
        'access_token': keycloak_access_token,
        'refresh_token': keycloak_refresh_token,
        'accepted_tos': accepted_tos,
    }
    from storage.encrypt_utils import get_jwt_service

    signed_token = get_jwt_service().create_jws_token(
        cookie_data, expires_in=timedelta(weeks=1)
    )

    # Set secure cookie with signed token. The value can exceed the
    # browser's 4096-byte single-cookie cap for users with large Keycloak
    # claim sets, so write it through the chunked-cookie helper...
    set_chunked_cookie(
        response,
        'keycloak_auth',
        signed_token,
        domain=get_cookie_domain(),
        secure=secure,
        httponly=True,
        samesite=get_cookie_samesite(),
    )

📄 enterprise/server/auth/cookie_chunking.py (第 1-52 行)

"""Chunked cookie helpers for the keycloak_auth session cookie.

Browsers cap a single cookie at ~4096 bytes (name + value + attributes).
The keycloak_auth cookie wraps a signed JWS containing the Keycloak
access and refresh tokens, and that can exceed the cap for users with
large claim sets (long emails, many realm roles, several allowed-origins
entries). Chrome silently drops an oversized cookie, which shows up as an
endless login loop...
"""

CHUNK_SIZE = 3000
MAX_CHUNKS = 8


def read_chunked_cookie(request: Request, key: str) -> str | None:
    """Reassemble a possibly-chunked cookie value, or None if absent.

    Concatenates key, key_1, key_2 ... in order, stopping at
    the first missing index. A plain single cookie reads back unchanged.
    """
    first = request.cookies.get(key)
    if first is None:
        return None
    parts = [first]
    for i in range(1, MAX_CHUNKS):
        part = request.cookies.get(_chunk_key(key, i))
        if part is None:
            break
        parts.append(part)
    return ''.join(parts)

为什么要分块？

Chrome 遇到超大 Cookie 会静默丢弃，不会报错。用户表现为"无限登录循环" — OAuth 回调成功但 Cookie 从未到达下一个请求。分块方案将值拆成 keycloak_auth、keycloak_auth_1、keycloak_auth_2 ... 最多 8 块（24KB），远大于任何实际 Token。

📄 enterprise/server/auth/authorization.py (第 48-108 行)

class Permission(str, Enum):
    """Permissions that can be assigned to roles."""

    # Secrets
    MANAGE_SECRETS = 'manage_secrets'

    # MCP
    MANAGE_MCP = 'manage_mcp'

    # Integrations
    MANAGE_INTEGRATIONS = 'manage_integrations'
    MANAGE_INTEGRATION_PROVIDERS = 'manage_integration_providers'

    # Application Settings
    MANAGE_APPLICATION_SETTINGS = 'manage_application_settings'

    # API Keys
    MANAGE_API_KEYS = 'manage_api_keys'

    # LLM Settings
    VIEW_LLM_SETTINGS = 'view_llm_settings'
    EDIT_LLM_SETTINGS = 'edit_llm_settings'

    # Billing
    VIEW_BILLING = 'view_billing'
    ADD_CREDITS = 'add_credits'

    # Organization Members
    INVITE_USER_TO_ORGANIZATION = 'invite_user_to_organization'
    CHANGE_USER_ROLE_MEMBER = 'change_user_role:member'
    CHANGE_USER_ROLE_ADMIN = 'change_user_role:admin'
    CHANGE_USER_ROLE_OWNER = 'change_user_role:owner'

    # Organization Management
    VIEW_ORG_SETTINGS = 'view_org_settings'
    CHANGE_ORGANIZATION_NAME = 'change_organization_name'
    DELETE_ORGANIZATION = 'delete_organization'
    CREATE_ORGANIZATION = 'create_organization'

    # Manage Automations
    MANAGE_AUTOMATIONS = 'manage_automations'

    # User provisioning
    PROVISION_USER = 'provision_user'

    # Organization Conversations (Admin/Owner only)
    VIEW_ORG_CONVERSATIONS = 'view_org_conversations'

    # Instance-level super-role administration
    MANAGE_SUPER_ADMINS = 'manage_super_admins'

📄 enterprise/server/auth/authorization.py (第 144-226 行)

# Base permission mappings for the regular (org-scoped) roles.
ROLE_PERMISSIONS: dict[RoleName, frozenset[Permission]] = {
    RoleName.OWNER: frozenset([
        # Full access to everything
        Permission.MANAGE_SECRETS,
        Permission.MANAGE_MCP,
        Permission.MANAGE_INTEGRATIONS,
        Permission.MANAGE_APPLICATION_SETTINGS,
        Permission.MANAGE_API_KEYS,
        Permission.VIEW_LLM_SETTINGS,
        Permission.EDIT_LLM_SETTINGS,
        Permission.VIEW_BILLING,
        Permission.ADD_CREDITS,
        Permission.INVITE_USER_TO_ORGANIZATION,
        Permission.CHANGE_USER_ROLE_MEMBER,
        Permission.CHANGE_USER_ROLE_ADMIN,
        Permission.CHANGE_USER_ROLE_OWNER,
        Permission.VIEW_ORG_SETTINGS,
        Permission.EDIT_ORG_SETTINGS,
        Permission.CHANGE_ORGANIZATION_NAME,
        Permission.DELETE_ORGANIZATION,
        Permission.MANAGE_ORG_CLAIMS,
        Permission.MANAGE_AUTOMATIONS,
        Permission.PROVISION_USER,
        Permission.VIEW_ORG_CONVERSATIONS,
        Permission.MANAGE_INTEGRATION_PROVIDERS,
    ]),
    RoleName.ADMIN: frozenset([
        # Almost everything except org deletion/rename
        Permission.MANAGE_SECRETS,
        Permission.MANAGE_MCP,
        Permission.MANAGE_INTEGRATIONS,
        ... (20 permissions total)
    ]),
    RoleName.MEMBER: frozenset([
        # Read + basic management
        Permission.MANAGE_SECRETS,
        Permission.MANAGE_MCP,
        Permission.MANAGE_INTEGRATIONS,
        Permission.MANAGE_APPLICATION_SETTINGS,
        Permission.MANAGE_API_KEYS,
        Permission.VIEW_ORG_SETTINGS,
        Permission.VIEW_LLM_SETTINGS,
        Permission.MANAGE_AUTOMATIONS,
    ]),
}

📄 enterprise/server/auth/authorization.py (第 233-248 行)

SUPER_ROLE_PERMISSIONS: dict[RoleName, frozenset[Permission]] = {
    RoleName.OWNER: frozenset(),
    RoleName.ADMIN: frozenset([
        Permission.CREATE_ORGANIZATION,
        Permission.PROVISION_USER,
        Permission.MANAGE_SUPER_ADMINS,
    ]),
    RoleName.MEMBER: frozenset(),
}

📄 enterprise/server/auth/authorization.py (第 337-383 行)

def has_permission(
    user_role: Role, permission: Permission, *, is_super: bool = False
) -> bool:
    if is_super:
        return permission in get_super_role_permissions(user_role.name)
    return permission in get_role_permissions(user_role.name)


async def authorize_permission(
    request: Request, user_id: str, permission: Permission
) -> None:
    """Enforce that user_id has permission in the request's target org.

    Raises HTTPException(403) otherwise. Mirrors the org-role then
    super-role fallback used by require_permission.
    """
    from server.auth.org_context import resolve_target_org_id_for_permission_check

    org_id = await resolve_target_org_id_for_permission_check(request)
    user_role = await get_user_org_role(user_id, org_id)
    if user_role and has_permission(user_role, permission):
        return
    super_role = await get_user_super_role(user_id)
    if super_role and has_permission(super_role, permission, is_super=True):
        return
    raise HTTPException(
        status_code=status.HTTP_403_FORBIDDEN,
        detail=f'Missing required permission: {permission.value}',
    )

📄 enterprise/storage/user_authorization.py (第 11-46 行)

class UserAuthorizationType(str, Enum):
    """Type of user authorization rule."""
    WHITELIST = 'whitelist'
    BLACKLIST = 'blacklist'


class UserAuthorization(Base):
    """Stores user authorization rules based on email patterns
    and provider types.

    Supports:
    - Email pattern matching using SQL LIKE (e.g., '%@openhands.dev')
    - Provider type filtering (e.g., 'github', 'gitlab')
    - Whitelist/Blacklist rules

    When email_pattern is NULL, the rule matches all emails.
    When provider_type is NULL, the rule matches all providers.
    """
    __tablename__ = 'user_authorizations'
    id: Mapped[int] = mapped_column(Identity(), primary_key=True)
    email_pattern: Mapped[str | None] = mapped_column(String, nullable=True)
    provider_type: Mapped[str | None] = mapped_column(String, nullable=True)
    type: Mapped[str] = mapped_column(String, nullable=False)
    created_at: Mapped[datetime] = mapped_column(...)
    updated_at: Mapped[datetime] = mapped_column(...)

📄 enterprise/server/auth/user/default_user_authorizer.py (第 32-79 行)

async def authorize_user(
    self, user_info: KeycloakUserInfo
) -> UserAuthorizationResponse:
    user_id = user_info.sub
    email = user_info.email
    provider_type = user_info.identity_provider
    try:
        if not email:
            return UserAuthorizationResponse(
                success=False, error_detail='missing_email'
            )

        if self.prevent_duplicates:
            has_duplicate = await token_manager.check_duplicate_base_email(
                email, user_id
            )
            if has_duplicate:
                return UserAuthorizationResponse(
                    success=False, error_detail='duplicate_email'
                )

        # Check authorization rules (whitelist takes precedence)
        base_email = extract_base_email(email)
        auth_type = await UserAuthorizationStore.get_authorization_type(
            base_email, provider_type
        )

        if auth_type == UserAuthorizationType.WHITELIST:
            return UserAuthorizationResponse(success=True)

        if auth_type == UserAuthorizationType.BLACKLIST:
            return UserAuthorizationResponse(success=False, error_detail='blocked')

📄 openhands/app_server/sandbox/session_auth.py (第 37-100 行)

async def validate_session_key(session_api_key: str | None) -> SandboxInfo:
    """Validate an X-Session-API-Key and return the associated sandbox.

    Security:
        This function enforces that session API keys are only valid for RUNNING
        sandboxes. This is a critical security measure to prevent leaked keys
        from being used to access user secrets after a sandbox has been paused,
        stopped, or deleted.

    Raises:
        HTTPException(401): if the key is missing or does not map to a sandbox.
        HTTPException(401): if the sandbox is not in RUNNING state.
        HTTPException(401): in SAAS mode if the sandbox has no owning user.
    """
    if not session_api_key:
        raise HTTPException(
            status.HTTP_401_UNAUTHORIZED,
            detail='X-Session-API-Key header is required',
        )

    # Use admin context to look up sandbox by session key
    state = InjectorState()
    setattr(state, USER_CONTEXT_ATTR, ADMIN)

    async with get_sandbox_service(state) as sandbox_service:
        sandbox_info = await sandbox_service.get_sandbox_by_session_api_key(
            session_api_key
        )

    if sandbox_info is None:
        raise HTTPException(
            status.HTTP_401_UNAUTHORIZED, detail='Invalid session API key'
        )

    # Security: Reject session keys for non-running sandboxes.
    if sandbox_info.status != SandboxStatus.RUNNING:
        raise HTTPException(
            status.HTTP_401_UNAUTHORIZED,
            detail='Sandbox is not running',
        )

    return sandbox_info

安全设计亮点：

Session API Key 只对 RUNNING 状态的沙箱有效。沙箱暂停、停止或删除后，即使 Key 泄露也无法访问用户密钥。这是一个"最小暴露窗口"的安全策略。

📄 enterprise/server/auth/auth_error.py (第 1-46 行)

class AuthError(Exception):
    """Generic auth error"""
    pass

class NoCredentialsError(AuthError):
    """Error when no authentication was provided"""
    pass

class EmailNotVerifiedError(AuthError):
    """Error when email is not verified"""
    pass

class BearerTokenError(AuthError):
    """Error when decoding a bearer token"""
    pass

class CookieError(AuthError):
    """Error when decoding an auth cookie"""
    pass

class TosNotAcceptedError(AuthError):
    """Error when decoding an auth cookie"""
    pass

class ExpiredError(AuthError):
    """Error when a token has expired (Usually the refresh token)"""
    pass

class TokenRefreshError(AuthError):
    """Error when token refresh fails due to timeout or lock contention"""
    pass