Server Authentication

Overview

Server Authentication manages credential verification, JWT lifecycle management, API key issuance, and session routing for the self-hosted FastAPI application and Next.js admin dashboard. It implements bcrypt password hashing with timing attack mitigation, secure access and refresh token generation, atomic refresh token rotation via database jti tracking, and multi-scheme request resolution supporting Bearer tokens, custom X-API-Key headers, and legacy administrator keys.

Sources: server/auth.py:11-165, server/routers/auth.py:94-165

Authentication Architecture and Credential Verification

Credential Hashing and Timing Mitigation

The server utilizes passlib with a CryptContext configured for the bcrypt hashing scheme with automatic deprecation handling. Passwords and API keys are processed through explicit hashing and verification routines.

Copypwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
 
def hash_password(password: str) -> str:
    return pwd_context.hash(password)
 
def verify_password(plain: str, hashed: str) -> bool:
    return pwd_context.verify(plain, hashed)

Sources: server/auth.py:22-30

To prevent email enumeration via timing attacks during login requests where a target email address does not exist in the database, the authentication flow invokes a dummy password verification method that executes identical bcrypt computation cycles.

Copydef dummy_verify_password() -> None:
    """Burn the same bcrypt cycles as a real verify so login timing doesn't leak whether an email exists."""
    pwd_context.dummy_verify()

Sources: server/auth.py:33-36

User Resolution Mechanisms

Incoming requests can be authenticated via three distinct mechanisms: JSON Web Tokens supplied via Bearer authorization headers, custom API keys transmitted in the X-API-Key header, or legacy admin keys. User resolution is decoupled into specific helper functions operating on isolated database sessions.

Sources: server/auth.py:144-175

JWT verification extracts the payload, validates that the token type is explicitly set to access, and queries the database for the corresponding user ID stored in the subject (sub) claim.

Copydef _resolve_user_from_jwt(token: str, db: Session) -> User:
    payload = decode_token(token)
    if payload.get("type") != "access":
        raise HTTPException(status_code=401, detail="Invalid token type.")
    user = db.get(User, payload.get("sub"))
    if user is None:
        raise HTTPException(status_code=401, detail="User not found.")
    return user

Sources: server/auth.py:116-123

API key resolution inspects the first 12 characters of the supplied key to isolate candidate records matching the stored key_prefix. It then iterates through candidates to verify the plaintext key against stored hashes using verify_api_key_hash, updating last_used_at upon a successful match.

Copydef _resolve_user_from_api_key(key: str, db: Session) -> User:
    prefix = key[:12] if len(key) >= 12 else key
    candidates = (
        db.execute(select(APIKey).where(APIKey.key_prefix == prefix, APIKey.revoked_at.is_(None))).scalars().all()
    )
 
    for candidate in candidates:
        if verify_api_key_hash(key, candidate.key_hash):
            candidate.last_used_at = datetime.now(timezone.utc)
            db.commit()
            user = db.get(User, candidate.created_by)
            if user is None:
                raise HTTPException(status_code=401, detail="API key owner not found.")
            return user
 
    raise HTTPException(status_code=401, detail="Invalid API key.")

Sources: server/auth.py:126-141

User Registration and Setup Status

Setup Status Inspection and Password Policies

The initial deployment workflow relies on determining whether an administrator account has been provisioned. The server exposes a setup status route configured at /auth/setup-status via the dashboard API endpoint collection, querying the database row count for User records. If the user count is precisely zero, the server responds with a boolean indicator signaling that setup is required.

Copy@router.get("/setup-status", response_model=SetupStatusResponse)
def setup_status(db: Session = Depends(get_db)):
    count = db.scalar(select(func.count(User.id)))
    return SetupStatusResponse(needsSetup=count == 0)

Sources: server/routers/auth.py:88-91, server/dashboard/src/utils/api-endpoints.ts:2

Password length enforcement is handled uniformly across registration and password modification paths. A mandatory validation helper inspects plain text strings against a fixed constant defining the minimum required character length.

Sources: server/routers/auth.py:28-36

Admin Account Registration and Telemetry Execution

Admin registration is permanently locked down after the first account is successfully persisted. The registration endpoint checks if any user already exists, hashes the verified password, assigns the "admin" role, and commits the transaction. To prevent race conditions from concurrent registration requests, database integrity errors trigger a rollback and raise a forbidden exception.

Copy@router.post("/register", response_model=TokenResponse)
@limiter.limit("5/minute")
def register(request: Request, body: RegisterRequest, db: Session = Depends(get_db)):
    """Create the first admin account. Blocked once any user exists."""
    _require_password_length(body.password)
 
    if db.scalar(select(func.count(User.id))) > 0:
        raise HTTPException(status_code=403, detail="Registration is closed. An admin account already exists.")
 
    user = User(
        name=body.name,
        email=body.email,
        password_hash=hash_password(body.password),
        role="admin",
    )
    db.add(user)
    try:
        db.commit()
    except IntegrityError:
        db.rollback()
        raise HTTPException(status_code=403, detail="Registration is closed. An admin account already exists.")
    db.refresh(user)
 
    capture_admin_registered(email=body.email)
 
    return TokenResponse(
        access_token=create_access_token(str(user.id), user.role),
        refresh_token=create_refresh_token(str(user.id), db),
    )

Sources: server/routers/auth.py:94-122

Upon successful creation of the administrator account, the execution flow triggers anonymous telemetry reporting via a specific function sequence. This sequence captures the event state, loads persistent state settings, generates installation identifiers, and initializes the PostHog client interface.

1. register — Invokes capture_admin_registered(email=body.email) after committing the new user to the database.
2. capture_admin_registered — Passes arguments to _capture_once(email, "admin_registered", "admin_registered_sent_at").
3. _capture_once — Verifies telemetry enablement, thread locks state access, and checks if the event was previously dispatched.
4. _get_client — Resolves or instantiates the PostHog tracking client using PROJECT_API_KEY and HOST.
5. _load_state — Reads persistent metrics configuration from STATE_PATH.
6. _install_id — Generates or retrieves the unique installation UUID.
7. _save_state — Serializes updated telemetry status back to disk.

Sources: server/routers/auth.py:117, server/telemetry.py:35-109

JWT Token Generation and Invalidation

Access and Refresh Token Issuance

Authentication sessions utilize JSON Web Tokens encoded using the HS256 algorithm with a secret key retrieved via _get_secret(). If JWT_SECRET is unconfigured, the helper raises an HTTP 500 error. Access tokens carry claims for the user ID (sub), role, expiration timestamp (exp), and a type identifier set to "access", with a fixed lifetime defined by ACCESS_TOKEN_EXPIRE_MINUTES.

Copydef create_access_token(user_id: str, role: str) -> str:
    expire = datetime.now(timezone.utc) + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    payload = {"sub": user_id, "role": role, "exp": expire, "type": "access"}
    return jwt.encode(payload, _get_secret(), algorithm=JWT_ALGORITHM)

Sources: server/auth.py:51-60

Refresh tokens operate with an extended expiration window governed by REFRESH_TOKEN_EXPIRE_DAYS and incorporate a unique JWT ID (jti) generated as a UUID. When a refresh token is created, a corresponding record is persisted in the database via the RefreshTokenJti model to track its usage and expiration state.

Copydef create_refresh_token(user_id: str, db: Session) -> str:
    expire = datetime.now(timezone.utc) + timedelta(days=REFRESH_TOKEN_EXPIRE_DAYS)
    jti = uuid.uuid4()
    db.add(RefreshTokenJti(jti=jti, user_id=uuid.UUID(user_id), expires_at=expire))
    db.commit()
    payload = {"sub": user_id, "exp": expire, "jti": str(jti), "type": "refresh"}
    return jwt.encode(payload, _get_secret(), algorithm=JWT_ALGORITHM)

Sources: server/auth.py:63-70

Refresh Token JTI Database Schema

The persistence layer for single-use refresh token tracking is defined by the refresh_token_jtis table created in Alembic migration version 005. The table stores the primary key jti, a cascade-deleting foreign key to users.id, expiration timestamps with timezone support, and a used_at timestamp marking consumption.

Copydef upgrade() -> None:
    op.create_table(
        "refresh_token_jtis",
        sa.Column("jti", sa.Uuid(), primary_key=True),
        sa.Column(
            "user_id",
            sa.Uuid(),
            sa.ForeignKey("users.id", ondelete="CASCADE"),
            nullable=False,
        ),
        sa.Column("expires_at", sa.DateTime(timezone=True), nullable=False),
        sa.Column("used_at", sa.DateTime(timezone=True), nullable=True),
        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()),
    )
    op.create_index("ix_refresh_token_jtis_expires_at", "refresh_token_jtis", ["expires_at"])
 

Sources: server/alembic/versions/005_refresh_token_jtis.py:20-35

Single-Use Token Rotation and Concurrency Control

To prevent token replay attacks and enforce single-use mechanics, refresh tokens are consumed atomically using a conditional database update. The consume_refresh_jti function attempts to set used_at only on rows where the jti matches, used_at is currently null, and expires_at exceeds the current UTC time.

Copydef consume_refresh_jti(jti: str, db: Session) -> None:
    try:
        jti_uuid = uuid.UUID(jti)
    except (TypeError, ValueError):
        raise HTTPException(status_code=401, detail="Refresh token is no longer valid.")
    now = datetime.now(timezone.utc)
    result = db.execute(
        update(RefreshTokenJti)
        .where(
            RefreshTokenJti.jti == jti_uuid,
            RefreshTokenJti.used_at.is_(None),
            RefreshTokenJti.expires_at > now,
        )
        .values(used_at=now)
    )
    if result.rowcount == 0:
        raise HTTPException(status_code=401, detail="Refresh token is no longer valid.")
    db.commit()

Sources: server/auth.py:72-94

The token refresh router endpoint validates the incoming refresh token payload type, extracts the jti claim, checks user existence, invokes consume_refresh_jti, and issues a brand new access and refresh token pair upon success.

Copy@router.post("/refresh", response_model=TokenResponse)
@limiter.limit("20/minute")
def refresh(request: Request, body: RefreshRequest, db: Session = Depends(get_db)):
    payload = decode_token(body.refresh_token)
    if payload.get("type") != "refresh":
        raise HTTPException(status_code=401, detail="Invalid token type.")
 
    jti = payload.get("jti")
    if not jti:
        raise HTTPException(status_code=401, detail="Refresh token is no longer valid.")
 
    user = db.get(User, payload["sub"])
    if user is None:
        raise HTTPException(status_code=401, detail="User not found.")
 
    consume_refresh_jti(jti, db)
 
    return TokenResponse(
        access_token=create_access_token(str(user.id), user.role),
        refresh_token=create_refresh_token(str(user.id), db),
    )

Sources: server/routers/auth.py:144-164

API Key Generation and Resolution

Key Generation and Hashing Mechanics

The server generates new API keys using cryptographic randomness combined with a fixed string prefix for identification. The generate_api_key function generates 32 bytes of URL-safe random data via secrets.token_urlsafe, prepends the m0sk_ format identifier, extracts a 12-character prefix for database indexing, and computes a secure bcrypt hash of the full plaintext key using CryptContext.

Copydef generate_api_key() -> tuple[str, str, str]:
    """Returns (full_key, prefix, hash)."""
    raw = secrets.token_urlsafe(32)
    full_key = f"m0sk_{raw}"
    prefix = full_key[:12]
    key_hash = pwd_context.hash(full_key)
    return full_key, prefix, key_hash

Sources: server/auth.py:38-44

Sources: server/auth.py:38-44

Request Header Resolution and Authentication Flow

Incoming requests can authenticate through multiple headers inspected by verify_auth and dependency helpers. The server reads credentials from either the Authorization: Bearer <token> header via bearer_scheme or the X-API-Key header via api_key_header.

Copybearer_scheme = HTTPBearer(auto_error=False)
api_key_header = APIKeyHeader(name="X-API-Key", auto_error=False)

Sources: server/auth.py:104-105

When an X-API-Key header is present, the resolver checks against the global environment variable ADMIN_API_KEY using constant-time comparison secrets.compare_digest. If it does not match the admin override, it queries the database for active records matching the key's 12-character prefix where revoked_at is null. It then verifies each candidate hash, updates last_used_at upon success, and resolves the owning user.

Copydef _resolve_user_from_api_key(key: str, db: Session) -> User:
    prefix = key[:12] if len(key) >= 12 else key
    candidates = (
        db.execute(select(APIKey).where(APIKey.key_prefix == prefix, APIKey.revoked_at.is_(None))).scalars().all()
    )
 
    for candidate in candidates:
        if verify_api_key_hash(key, candidate.key_hash):
            candidate.last_used_at = datetime.now(timezone.utc)
            db.commit()
            user = db.get(User, candidate.created_by)
            if user is None:
                raise HTTPException(status_code=401, detail="API key owner not found.")
            return user
 
    raise HTTPException(status_code=401, detail="Invalid API key.")

Sources: server/auth.py:126-141

Dashboard Session Refresh and Middleware

Session Cookie Rotation and Refresh Route

The Next.js dashboard manages authentication persistence via an HTTP-only cookie containing the refresh token (mem0_refresh_token). The server route handler exposes POST, PUT, and DELETE methods at /api/auth/refresh to rotate tokens, update sessions, and clear credentials.

Copyconst COOKIE_NAME = "mem0_refresh_token";
 
const COOKIE_OPTIONS = {
  httpOnly: true,
  secure: shouldUseSecureCookie(),
  sameSite: "lax" as const,
  path: "/",
  maxAge: 30 * 24 * 60 * 60, // 30 days
};

Sources: server/dashboard/src/app/api/auth/refresh/route.ts:6-27

When a POST request is received, the route extracts the refresh token from the cookie store and forwards it to the backend server endpoint. If the backend accepts the token, a new refresh token is stored in the cookie, and a short-lived access token is returned in the response body. If verification fails, the refresh token cookie is deleted and a 401 status is returned.

Sources: server/dashboard/src/app/api/auth/refresh/route.ts:29-75

Next.js Middleware Token Routing and Setup Enforcements

The Next.js middleware intercepts incoming requests across the application, bypassing static assets and authentication endpoints defined in PUBLIC_PATHS. It inspects the presence of the mem0_refresh_token cookie and queries the backend setup status when users visit root or setup routes.

Copyconst PUBLIC_PATHS = [
  "/_next",
  "/api/auth",
  "/api/health",
  "/fonts",
  "/favicon",
];

Sources: server/dashboard/src/middleware.ts:5-11

The matcher configuration excludes static assets and icons from middleware evaluation, while routing protected navigation targets appropriately.

Copyexport const config = {
  matcher: ["/((?!_next/static|_next/image|favicon.ico|fonts|images|icons).*)"],
};

Sources: server/dashboard/src/middleware.ts:65-67

Administrative Password Reset Script

Utility Execution and Credential Updates

The administrative password reset utility is designed to run directly inside the running container environment via Python. It reads the target user's EMAIL and new PASSWORD values directly from environment variables. Sources: server/scripts/reset_admin_password.py:1-17

CopyMIN_PASSWORD_LENGTH = 8

Sources: server/scripts/reset_admin_password.py:12-12

The script opens a database session using SessionLocal, queries the User model by email address, computes a fresh bcrypt password hash via hash_password, and commits the updated credential to the database. Sources: server/scripts/reset_admin_password.py:27-35

Sources: server/scripts/reset_admin_password.py:19-37

Makefile Integration

Container orchestration commands expose the reset script through a dedicated reset-admin-password Makefile target. This target validates that both EMAIL and PASSWORD parameters are supplied before executing the command inside the docker compose container environment. Sources: server/Makefile:50-54

Copyreset-admin-password:
	@test -n "$(EMAIL)" || (echo "usage: make reset-admin-password EMAIL=<email> PASSWORD=<password>" && exit 2)
	@test -n "$(PASSWORD)" || (echo "usage: make reset-admin-password EMAIL=<email> PASSWORD=<password>" && exit 2)
	@docker compose exec -T -e EMAIL="$(EMAIL)" -e PASSWORD="$(PASSWORD)" -e PYTHONPATH=/app mem0 python scripts/reset_admin_password.py

Sources: server/Makefile:50-54

Related

• FastAPI Application
• Database and Migrations