Repository: marimo-team/marimo
Pull request: feat: basic-auth support and minimal login page
Fixes #395

This adds basic auth support. It is both _basic_ and uses the `Basic` auth schema.

Overview:
* it is enabled by default for `marimo edit/tutorial/new`, and not enabled by default for `marimo run`
* You can enable/disable with `--token/--no-token` and `--token-password` to customize the password
* Rename `ValidateServerTokenMiddleware` to `SkewProtectionMiddleware`

---

In order to authenticate, you must either pass the token as a password in the `Authorization` header, or as a query parameter under `access_token` in the URL.

1. Basic Authorization header:

To authenticate using the `Authorization` header, you must pass the token as a password in the `Authorization` header using the [Basic authentication scheme](https://developer.mozilla.org/en-US/docs/Web/HTTP/Authentication). For example, to authenticate with the token `sup3rs3cr3t`, you would pass the header `Authorization Basic base64("any_username:sup3rs3cr3t")`.

This is not necessary when using a browser, as we will redirect you to a minimal login page

2. Query parameter (most common flow):

To authenticate using a query parameter, you must pass the token as a query parameter under `access_token` in the URL. For example, to authenticate with the token `sup3rs3cr3t`, you would pass the query parameter `http://localhost:2718?access_token=sup3rs3cr3t`.

For convenience, when running locally, marimo will automatically open the URL with the query parameter in your default browser.


Problem statement:
password required when logging in
### Description

hi 👋, I just started trying marimo.

I deployed this as a public service on the server, considering that there might be some security risks. I'm hoping to restrict logins in some way.

Not sure if I've missed something, if marimo can do this please point it out.

Thank you for your great work!

### Suggested solution

If a password is set in the configuration, you need to provide the password to enter marimo.

### Alternative

_No response_

### Additional context

_No response_

Hints from issue discussion:
Hi @jetjinser! Thanks for the great suggestion.

marimo doesn't currently support password authentication -- sorry about that! But we can definitely add this to our list of future features. 
@jetjinser  - you should be able to roll this yourself using `marimo.create_asgi_app`.

This example doesn't implement the password protection, but it close. you can likely go off this: https://github.com/marimo-team/marimo/blob/main/marimo/_smoke_tests/custom_server/my_server.py
FastAPI provides some middlewares and utils for sessions to make this even easier. 

Here are some docs: https://docs.marimo.io/guides/deploying/programmatically.html
Going to re-open this as we plan to add basic-auth support: https://github.com/marimo-team/marimo/pull/1363

Commit message: delete access_token, remove from query_params

Patch to review:
diff --git a/docs/guides/authentication.md b/docs/guides/authentication.md
new file mode 100644
index 00000000000..a1fac934421
--- /dev/null
+++ b/docs/guides/authentication.md
@@ -0,0 +1,67 @@
+# Authentication
+
+marimo provides a simple way to add token/password protection to your marimo server. Given that authentication is a complex topic, marimo does not provide a built-in authentication/authorization system, but instead makes it easy to add your own through ASGI middleware.
+
+## Enabling Basic Authentication
+
+Authentication is disabled by default, however this may change in the future. To enable authentication, you must pass `--token` to your `marimo edit/run/new` command from the Terminal. This will use a randomly generated token in `Edit mode` and a deterministic token in `Run mode` (based on the code of the notebook).
+
+```bash
+marimo run my_notebook.py --auth --token="sup3rs3cr3t"
+```
+
+### Ways to Authenticate
+
+In order to authenticate, you must either pass the token as a password in the `Authorization` header, or as a query parameter under `access_token` in the URL.
+
+1. Basic Authorization header:
+
+To authenticate using the `Authorization` header, you must pass the token as a password in the `Authorization` header using the [Basic authentication scheme](https://developer.mozilla.org/en-US/docs/Web/HTTP/Authentication). For example, to authenticate with the token `sup3rs3cr3t`, you would pass the header `Authorization Basic base64("any_username:sup3rs3cr3t")`.
+
+This is not necessary when using a browser, as the browser will automatically prompt you for a username and password. Note: any username can be used.
+
+2. Query parameter:
+
+To authenticate using a query parameter, you must pass the token as a query parameter under `access_token` in the URL. For example, to authenticate with the token `sup3rs3cr3t`, you would pass the query parameter `http://localhost:2718?access_token=sup3rs3cr3t`.
+
+For convenience, when running locally, marimo will automatically open the URL with the query parameter in your default browser.
+
+## Custom Authentication
+
+If you choose to make your marimo application public, you may want to add your own authentication system, along with authorization, rate limiting, etc. You can do this by creating a marimo application programmatically and adding your own middleware to the ASGI application.
+
+Here's an example of how you can add authentication to a marimo application using FastAPI:
+
+```python
+from typing import Annotated, Callable, Coroutine
+from fastapi.responses import HTMLResponse, RedirectResponse
+import marimo
+from fastapi import FastAPI, Form, Request, Response
+# Custom auth middleware and login page
+from my_auth_module import auth_middleware, my_login_route
+
+
+# Create a marimo asgi app
+server = (
+    marimo.create_asgi_app()
+    .with_app(path="", root="./pages/index.py")
+    .with_app(path="/dashboard", root="./pages/dashboard.py")
+    .with_app(path="/sales", root="./pages/sales.py")
+)
+
+# Create a FastAPI app
+app = FastAPI()
+
+app.add_middleware(auth_middleware)
+app.add_route("/login", my_login_route, methods=["POST"])
+
+app.mount("/", server.build())
+
+# Run the server
+if __name__ == "__main__":
+    import uvicorn
+
+    uvicorn.run(app, host="localhost", port=8000)
+```
+
+or for a full example on implementing OAuth2 with FastAPI, see the [FastAPI OAuth2 example](https://fastapi.tiangolo.com/tutorial/security/oauth2-jwt/).
diff --git a/docs/guides/index.md b/docs/guides/index.md
index 29dfd4e97f6..ccabf75da01 100644
--- a/docs/guides/index.md
+++ b/docs/guides/index.md
@@ -22,6 +22,7 @@
   ai_completion
   best_practices
   performance
+  authentication
   exporting
   deploying/index
 ```
@@ -48,6 +49,7 @@ These guides cover marimo's core concepts.
 | {doc}`wasm`                          | Running notebooks in the browser (no backend required!) |
 | {doc}`ai_completion`                 | Using AI to speed up your coding                        |
 | {doc}`exporting`                     | Exporting notebooks to HTML and flat scripts            |
+| {doc}`authentication`                | Authentication and security                             |
 | {doc}`deploying/index`               | Deploying marimo notebooks and apps                     |
 
 ```{admonition} Learn by doing!
diff --git a/frontend/playwright.config.ts b/frontend/playwright.config.ts
index d0f614a6fd5..88a600a4225 100644
--- a/frontend/playwright.config.ts
+++ b/frontend/playwright.config.ts
@@ -176,7 +176,7 @@ const config: PlaywrightTestConfig = {
       const baseUrl = options.command === "run" ? options.baseUrl : undefined;
 
       const pathToApp = path.join(pydir, app);
-      let marimoCmd = `marimo -q ${command} ${pathToApp} -p ${port} --headless`;
+      let marimoCmd = `marimo -q ${command} ${pathToApp} -p ${port} --headless --no-token`;
       if (baseUrl) {
         marimoCmd += ` --base-url=${baseUrl}`;
       }
@@ -188,7 +188,7 @@ const config: PlaywrightTestConfig = {
       };
     }),
     {
-      command: `marimo -q edit -p ${EDIT_PORT} --headless`,
+      command: `marimo -q edit -p ${EDIT_PORT} --headless --no-token`,
       url: getUrl(EDIT_PORT),
       reuseExistingServer: false,
     },
diff --git a/frontend/src/core/network/auth.ts b/frontend/src/core/network/auth.ts
new file mode 100644
index 00000000000..0d496f76bfd
--- /dev/null
+++ b/frontend/src/core/network/auth.ts
@@ -0,0 +1,12 @@
+/* Copyright 2024 Marimo. All rights reserved. */
+
+import { updateQueryParams } from "@/utils/urls";
+
+/**
+ * Remove access_token from the query string.
+ */
+export function cleanupAuthQueryParams() {
+  updateQueryParams((params) => {
+    params.delete("access_token");
+  });
+}
diff --git a/frontend/src/main.tsx b/frontend/src/main.tsx
index 3e6fe10c487..86c08728cce 100644
--- a/frontend/src/main.tsx
+++ b/frontend/src/main.tsx
@@ -11,9 +11,11 @@ import { patchFetch, patchVegaLoader } from "./core/static/files";
 import { isStaticNotebook } from "./core/static/static-state";
 import { vegaLoader } from "./plugins/impl/vega/loader";
 import { initializePlugins } from "./plugins/plugins";
+import { cleanupAuthQueryParams } from "./core/network/auth";
 
 maybeRegisterVSCodeBindings();
 initializePlugins();
+cleanupAuthQueryParams();
 
 /**
  * Main entry point for the Marimo app.
diff --git a/marimo/_cli/cli.py b/marimo/_cli/cli.py
index 5d20a1d4ec1..bb618670737 100644
--- a/marimo/_cli/cli.py
+++ b/marimo/_cli/cli.py
@@ -23,6 +23,7 @@
 from marimo._server.file_router import AppFileRouter
 from marimo._server.model import SessionMode
 from marimo._server.start import start
+from marimo._server.tokens import AuthToken
 from marimo._utils.marimo_path import MarimoPath
 
 DEVELOPMENT_MODE = False
@@ -66,6 +67,18 @@ def _sep(desc: str) -> str:
     return "\n".join(lines)
 
 
+def _resolve_token(
+    token: bool, token_password: Optional[str]
+) -> Optional[AuthToken]:
+    if token_password:
+        return AuthToken(token_password)
+    elif token is False:
+        # Empty means no auth
+        return AuthToken("")
+    # None means use the default (generated) token
+    return None
+
+
 main_help_msg = "\n".join(
     [
         "\b",
@@ -102,6 +115,20 @@ def _sep(desc: str) -> str:
     ]
 )
 
+token_message = """
+    Use a token for authentication.
+    This enables session-based authentication.
+    A random token will be generated if --token-password is not set.
+
+    If --no-token is set, session-based authentication will not be used.
+    """
+
+token_password_message = """
+    Use a specific token for authentication.
+    This enables session-based authentication.
+    A random token will be generated if not set.
+    """
+
 
 @click.group(help=main_help_msg)
 @click.version_option(version=__version__, message="%(version)s")
@@ -182,12 +209,28 @@ def main(log_level: str, quiet: bool, development_mode: bool) -> None:
     type=bool,
     help="Don't launch a browser.",
 )
+@click.option(
+    "--token/--no-token",
+    default=True,
+    show_default=True,
+    type=bool,
+    help=token_message,
+)
+@click.option(
+    "--token-password",
+    default=None,
+    show_default=True,
+    type=str,
+    help=token_password_message,
+)
 @click.argument("name", required=False)
 @click.argument("args", nargs=-1, type=click.UNPROCESSED)
 def edit(
     port: Optional[int],
     host: str,
     headless: bool,
+    token: bool,
+    token_password: Optional[str],
     name: Optional[str],
     args: tuple[str],
 ) -> None:
@@ -228,6 +271,7 @@ def edit(
         include_code=True,
         watch=False,
         cli_args=parse_args(args),
+        auth_token=_resolve_token(token, token_password),
     )
 
 
@@ -255,10 +299,26 @@ def edit(
     type=bool,
     help="Don't launch a browser.",
 )
+@click.option(
+    "--token/--no-token",
+    default=True,
+    show_default=True,
+    type=bool,
+    help=token_message,
+)
+@click.option(
+    "--token-password",
+    default=None,
+    show_default=True,
+    type=str,
+    help=token_password_message,
+)
 def new(
     port: Optional[int],
     host: str,
     headless: bool,
+    token: bool,
+    token_password: Optional[str],
 ) -> None:
     start(
         file_router=AppFileRouter.new_file(),
@@ -271,6 +331,7 @@ def new(
         include_code=True,
         watch=False,
         cli_args={},
+        auth_token=_resolve_token(token, token_password),
     )
 
 
@@ -308,6 +369,20 @@ def new(
     type=bool,
     help="Don't launch a browser.",
 )
+@click.option(
+    "--token/--no-token",
+    default=False,
+    show_default=True,
+    type=bool,
+    help=token_message,
+)
+@click.option(
+    "--token-password",
+    default=None,
+    show_default=True,
+    type=str,
+    help=token_password_message,
+)
 @click.option(
     "--include-code",
     is_flag=True,
@@ -342,6 +417,8 @@ def run(
     port: Optional[int],
     host: str,
     headless: bool,
+    token: bool,
+    token_password: Optional[str],
     include_code: bool,
     watch: bool,
     base_url: str,
@@ -369,6 +446,7 @@ def run(
         watch=watch,
         base_url=base_url,
         cli_args=parse_args(args),
+        auth_token=_resolve_token(token, token_password),
     )
 
 
@@ -431,6 +509,20 @@ def recover(name: str) -> None:
     type=bool,
     help="Don't launch a browser.",
 )
+@click.option(
+    "--token/--no-token",
+    default=True,
+    show_default=True,
+    type=bool,
+    help=token_message,
+)
+@click.option(
+    "--token-password",
+    default=None,
+    show_default=True,
+    type=str,
+    help=token_password_message,
+)
 @click.argument(
     "name",
     required=True,
@@ -451,6 +543,8 @@ def tutorial(
     port: Optional[int],
     host: str,
     headless: bool,
+    token: bool,
+    token_password: Optional[str],
     name: Literal[
         "intro",
         "dataflow",
@@ -500,6 +594,7 @@ def tutorial(
         headless=headless,
         watch=False,
         cli_args={},
+        auth_token=_resolve_token(token, token_password),
     )
 
 
diff --git a/marimo/_runtime/params.py b/marimo/_runtime/params.py
index 61e9303ce19..f53614e8ca5 100644
--- a/marimo/_runtime/params.py
+++ b/marimo/_runtime/params.py
@@ -25,6 +25,8 @@
 class QueryParams(State[SerializedQueryParams]):
     """Query parameters for a marimo app."""
 
+    IGNORED_KEYS = {"access_token", "refresh_token", "session_id"}
+
     def __init__(
         self,
         params: Dict[str, Union[str, List[str]]],
diff --git a/marimo/_server/api/auth.py b/marimo/_server/api/auth.py
new file mode 100644
index 00000000000..cefdb4e0754
--- /dev/null
+++ b/marimo/_server/api/auth.py
@@ -0,0 +1,190 @@
+# Copyright 2024 Marimo. All rights reserved.
+from __future__ import annotations
+
+import base64
+import secrets
+import typing
+from typing import TYPE_CHECKING, Any, Dict, Optional
+
+import starlette.status as status
+from starlette.datastructures import Secret
+from starlette.exceptions import HTTPException
+from starlette.middleware.sessions import SessionMiddleware
+from starlette.responses import JSONResponse
+
+from marimo import _loggers
+
+if TYPE_CHECKING:
+    from starlette.authentication import AuthenticationError
+    from starlette.requests import HTTPConnection
+    from starlette.types import ASGIApp, Receive, Scope, Send
+
+from marimo._server.api.deps import AppState
+
+LOGGER = _loggers.marimo_logger()
+TOKEN_QUERY_PARAM = "access_token"
+
+
+# Validates auth
+# - Checking for existing session cookie (already authenticated)
+# - Or authenticates by access_token in query params
+# - Or authenticates by basic auth
+def validate_auth(conn: HTTPConnection) -> bool:
+    state = AppState.from_app(conn.app)
+    auth_token = str(state.session_manager.auth_token)
+
+    # Check for session cookie
+    cookie_session = CookieSession(conn.session)
+    # Validate the cookie
+    if cookie_session.get_access_token() == auth_token:
+        return True  # Success
+
+    # Check for access_token
+    if TOKEN_QUERY_PARAM in conn.query_params:
+        # Validate the access_token
+        if conn.query_params[TOKEN_QUERY_PARAM] == auth_token:
+            LOGGER.debug("Validated access_token")
+            # Set the cookie
+            cookie_session.set_access_token(auth_token)
+            return True  # Success
+
+    # Check for basic auth
+    auth = conn.headers.get("Authorization")
+    if auth is not None:
+        username, password = _parse_basic_auth_header(auth)
+        if username and password == auth_token:
+            LOGGER.debug("Validated basic auth")
+            # Set the cookie
+            cookie_session.set_access_token(auth_token)
+            cookie_session.set_username(username)
+            return True  # Success
+
+    LOGGER.debug("Invalid auth")
+    return False
+
+
+def _parse_basic_auth_header(
+    header: str,
+) -> tuple[Optional[str], Optional[str]]:
+    scheme, _, credentials = header.partition(" ")
+
+    if scheme.lower() != "basic":
+        LOGGER.debug("Invalid auth scheme: %s", scheme)
+        return None, None
+
+    decoded = base64.b64decode(credentials).decode("utf-8")
+    username, _, password = decoded.partition(":")
+
+    if not password:
+        return None, None
+
+    LOGGER.debug("Validated basic auth for user: %s", username)
+
+    return username, password
+
+
+def raise_basic_auth_error() -> HTTPException:
+    return HTTPException(
+        status_code=status.HTTP_401_UNAUTHORIZED,
+        detail="Authorization header required",
+        headers={"WWW-Authenticate": "Basic"},
+    )
+
+
+def on_auth_error(
+    request: HTTPConnection, error: AuthenticationError
+) -> JSONResponse:
+    del request
+    del error
+    return JSONResponse(
+        status_code=status.HTTP_401_UNAUTHORIZED,
+        content={"detail": "Authorization header required"},
+        headers={"WWW-Authenticate": "Basic"},
+    )
+
+
+# This is random/new for each server instance
+RANDOM_SECRET = Secret(secrets.token_hex(32))
+
+
+class CookieSession:
+    """
+    Wrapper around starlette's Session to add typesafety
+    """
+
+    def __init__(self, session_state: Dict[str, Any]) -> None:
+        self.session_state = session_state
+
+    def get_access_token(self) -> str:
+        access_token: str = self.session_state.get("access_token", "")
+        return access_token
+
+    def get_username(self) -> str:
+        username: str = self.session_state.get("username", "")
+        return username
+
+    def set_access_token(self, token: str) -> None:
+        self.session_state["access_token"] = token
+
+    def set_username(self, username: str) -> None:
+        self.session_state["username"] = username
+
+
+class CustomSessionMiddleware(SessionMiddleware):
+    """
+    Wrapper around starlette's SessionMiddleware to:
+     - customize the session cookie based on the the port
+     - only run in Edit mode
+    """
+
+    def __init__(
+        self,
+        app: ASGIApp,
+        secret_key: typing.Union[str, Secret],
+        session_cookie: str = "session",
+        max_age: typing.Optional[int] = 14
+        * 24
+        * 60
+        * 60,  # 14 days, in seconds
+        path: str = "/",
+        same_site: typing.Literal["lax", "strict", "none"] = "lax",
+        https_only: bool = False,
+        domain: typing.Optional[str] = None,
+    ) -> None:
+        # We can't update the cookie here since
+        # we don't have access to the app state
+
+        self.original_session_cookie = session_cookie
+
+        super().__init__(
+            app,
+            secret_key,
+            session_cookie,
+            max_age,
+            path,
+            same_site,
+            https_only,
+            domain,
+        )
+
+    async def __call__(
+        self, scope: Scope, receive: Receive, send: Send
+    ) -> None:
+        if scope["type"] not in ("http", "websocket"):
+            await self.app(scope, receive, send)
+            return
+
+        state = AppState.from_app(scope["app"])
+        # disabled = state.session_manager.mode != SessionMode.EDIT
+        # if disabled:
+        #     return await self.app(scope, receive, send)
+
+        # We key the token cookie by port to avoid conflicts
+        # with multiple marimo instances running on the same host
+        maybe_port = state.maybe_port
+        if maybe_port is not None:
+            self.session_cookie = (
+                f"{self.original_session_cookie}_{maybe_port}"
+            )
+
+        return await super().__call__(scope, receive, send)
diff --git a/marimo/_server/api/deps.py b/marimo/_server/api/deps.py
index ef7c4446372..d008dfeb401 100644
--- a/marimo/_server/api/deps.py
+++ b/marimo/_server/api/deps.py
@@ -1,75 +1,42 @@
 # Copyright 2024 Marimo. All rights reserved.
 from __future__ import annotations
 
-from typing import TYPE_CHECKING, Optional, Union
+from typing import TYPE_CHECKING, Any, Optional, Union, cast
 
 from marimo._config.manager import UserConfigManager
 from marimo._server.ids import SessionId
 from marimo._server.model import SessionMode
 from marimo._server.sessions import Session, SessionManager
+from marimo._server.tokens import SkewProtectionToken
 
 if TYPE_CHECKING:
+    from starlette.applications import Starlette
+    from starlette.datastructures import State
     from starlette.requests import Request
     from starlette.websockets import WebSocket
     from uvicorn import Server
 
 
-def app_state(request: Request) -> AppState:
-    """Get the app state."""
-    return AppState(request.app)
-
-
-class AppState:
+class AppStateBase:
     """The app state."""
 
-    def __init__(self, request: Union[Request, WebSocket]) -> None:
-        """Initialize the app state."""
-        self.request = request
-
-        assert (
-            request.app.state.session_manager is not None
-        ), "Session manager not initialized"
-        self.session_manager: SessionManager = (
-            request.app.state.session_manager
-        )
-
-        assert (
-            request.app.state.base_url is not None
-        ), "Base URL not initialized"
-        self._base_url: str = request.app.state.base_url
-
-        assert (
-            request.app.state.config_manager is not None
-        ), "Config manager not initialized"
-        self._config_manager: UserConfigManager = (
-            request.app.state.config_manager
-        )
-
-    def get_current_session_id(self) -> Optional[SessionId]:
-        """Get the current session."""
-        return self.request.headers.get("Marimo-Session-Id")
+    @staticmethod
+    def from_request(request: Union[Request, WebSocket]) -> AppState:
+        """Get the app state with a request."""
+        return AppState(request)
 
-    def require_current_session_id(self) -> SessionId:
-        """Get the current session or raise an error."""
-        session_id = self.get_current_session_id()
-        if session_id is None:
-            raise ValueError("Missing Marimo-Session-Id header")
-        return session_id
+    @staticmethod
+    def from_app(asgi: Starlette) -> AppStateBase:
+        """Get the app state with an ASGIApp app."""
+        return AppStateBase(cast(Any, asgi).state)
 
-    def get_current_session(self) -> Optional[Session]:
-        """Get the current session."""
-        session_id = self.get_current_session_id()
-        if session_id is None:
-            return None
-        return self.session_manager.get_session(session_id)
+    def __init__(self, state: State) -> None:
+        """Initialize the app state."""
+        self.state = state
 
-    def require_current_session(self) -> Session:
-        """Get the current session or raise an error."""
-        session_id = self.require_current_session_id()
-        session = self.session_manager.get_session(session_id)
-        if session is None:
-            raise ValueError(f"Invalid session id: {session_id}")
-        return session
+    @property
+    def session_manager(self) -> SessionManager:
+        return cast(SessionManager, self.state.session_manager)
 
     @property
     def mode(self) -> SessionMode:
@@ -85,35 +52,81 @@ def development_mode(self) -> bool:
 
     @property
     def host(self) -> str:
-        assert self.request.app.state.host is not None, "Host not initialized"
-        host: str = self.request.app.state.host
+        host: str = self.state.host
         return host
 
     @property
     def port(self) -> int:
-        assert self.request.app.state.port is not None, "Port not initialized"
-        post: int = self.request.app.state.port
+        post: int = self.state.port
         return post
 
     @property
-    def base_url(self) -> str:
-        return self._base_url
+    def maybe_port(self) -> Optional[int]:
+        return self.state._state.get("port")
 
     @property
-    def server_token(self) -> str:
-        return self.session_manager.server_token
+    def base_url(self) -> str:
+        base_url: str = self.state.base_url
+        return base_url
 
     @property
     def server(self) -> Server:
-        assert (
-            self.request.app.state.server is not None
-        ), "Server not initialized"
-        server: Server = self.request.app.state.server
+        server: Server = self.state.server
         return server
 
     @property
     def config_manager(self) -> UserConfigManager:
-        return self._config_manager
+        cm: UserConfigManager = self.state.config_manager
+        return cm
+
+    @property
+    def watch(self) -> bool:
+        watch: bool = self.state.watch
+        return watch
+
+    @property
+    def headless(self) -> bool:
+        headless: bool = self.state.headless
+        return headless
+
+    @property
+    def skew_protection_token(self) -> SkewProtectionToken:
+        return self.session_manager.skew_protection_token
+
+
+class AppState(AppStateBase):
+    """The app state with a request."""
+
+    def __init__(self, request: Union[Request, WebSocket]) -> None:
+        """Initialize the app state with a request."""
+        super().__init__(request.app.state)
+        self.request = request
+
+    def get_current_session_id(self) -> Optional[SessionId]:
+        """Get the current session."""
+        return self.request.headers.get("Marimo-Session-Id")
+
+    def require_current_session_id(self) -> SessionId:
+        """Get the current session or raise an error."""
+        session_id = self.get_current_session_id()
+        if session_id is None:
+            raise ValueError("Missing Marimo-Session-Id header")
+        return session_id
+
+    def get_current_session(self) -> Optional[Session]:
+        """Get the current session."""
+        session_id = self.get_current_session_id()
+        if session_id is None:
+            return None
+        return self.session_manager.get_session(session_id)
+
+    def require_current_session(self) -> Session:
+        """Get the current session or raise an error."""
+        session_id = self.require_current_session_id()
+        session = self.session_manager.get_session(session_id)
+        if session is None:
+            raise ValueError(f"Invalid session id: {session_id}")
+        return session
 
     def require_query_params(self, param: str) -> str:
         """Get a query parameter or raise an error."""
diff --git a/marimo/_server/api/endpoints/assets.py b/marimo/_server/api/endpoints/assets.py
index 44d1cbfa0af..a32764bb479 100644
--- a/marimo/_server/api/endpoints/assets.py
+++ b/marimo/_server/api/endpoints/assets.py
@@ -6,6 +6,7 @@
 import re
 from typing import TYPE_CHECKING
 
+from starlette.authentication import requires
 from starlette.exceptions import HTTPException
 from starlette.responses import FileResponse, HTMLResponse, Response
 from starlette.staticfiles import StaticFiles
@@ -47,6 +48,7 @@
 
 
 @router.get("/")
+@requires("read")
 async def index(request: Request) -> HTMLResponse:
     app_state = AppState(request)
     user_config = app_state.config_manager.get_config()
@@ -67,7 +69,7 @@ async def index(request: Request) -> HTMLResponse:
             html=html,
             base_url=app_state.base_url,
             user_config=user_config,
-            server_token=app_state.server_token,
+            server_token=app_state.skew_protection_token,
         )
     else:
         # We have a file key, so we can render the app with the file
@@ -79,7 +81,7 @@ async def index(request: Request) -> HTMLResponse:
             html=html,
             base_url=app_state.base_url,
             user_config=user_config,
-            server_token=app_state.server_token,
+            server_token=app_state.skew_protection_token,
             app_config=app_config,
             filename=app_manager.filename,
             mode=app_state.mode,
@@ -98,6 +100,7 @@ async def index(request: Request) -> HTMLResponse:
 
 
 @router.get("/@file/{filename_and_length:path}")
+@requires("read")
 def virtual_file(
     request: Request,
 ) -> Response:
diff --git a/marimo/_server/api/endpoints/health.py b/marimo/_server/api/endpoints/health.py
index 901404aa52f..1299eb07f88 100644
--- a/marimo/_server/api/endpoints/health.py
+++ b/marimo/_server/api/endpoints/health.py
@@ -31,6 +31,7 @@ async def health_check(request: Request) -> JSONResponse:
 
 
 @router.get("/api/status")
+@requires("edit")
 async def status(request: Request) -> JSONResponse:
     app_state = AppState(request)
     files = [
diff --git a/marimo/_server/api/endpoints/ws.py b/marimo/_server/api/endpoints/ws.py
index fd0eab45d14..62f5a42927b 100644
--- a/marimo/_server/api/endpoints/ws.py
+++ b/marimo/_server/api/endpoints/ws.py
@@ -287,7 +287,7 @@ async def get_session() -> Session:
             # session was created.
             query_params = QueryParams({}, NoopStream())
             for key, value in self.websocket.query_params.multi_items():
-                if SESSION_QUERY_PARAM_KEY == key:
+                if key in QueryParams.IGNORED_KEYS:
                     continue
                 query_params.append(key, value)
 
diff --git a/marimo/_server/api/lifespans.py b/marimo/_server/api/lifespans.py
index fb427b349e1..70e92d6e5bd 100644
--- a/marimo/_server/api/lifespans.py
+++ b/marimo/_server/api/lifespans.py
@@ -6,8 +6,10 @@
 import socket
 import sys
 
+from marimo._server.api.deps import AppState, AppStateBase
 from marimo._server.file_router import AppFileRouter
 from marimo._server.sessions import SessionManager
+from marimo._server.tokens import AuthToken
 
 if sys.version_info < (3, 9):
     from typing import (
@@ -72,8 +74,9 @@ def __call__(self, app: Starlette) -> AbstractAsyncContextManager[None]:
 
 @contextlib.asynccontextmanager
 async def lsp(app: Starlette) -> AsyncIterator[None]:
-    user_config = app.state.config_manager.get_config()
-    session_mgr = app.state.session_manager
+    state = AppState.from_app(app)
+    user_config = state.config_manager.get_config()
+    session_mgr = state.session_manager
     run = session_mgr.mode == SessionMode.RUN
     if not run and user_config["completion"]["copilot"]:
         LOGGER.debug("GitHub Copilot is enabled")
@@ -83,22 +86,19 @@ async def lsp(app: Starlette) -> AsyncIterator[None]:
 
 @contextlib.asynccontextmanager
 async def watcher(app: Starlette) -> AsyncIterator[None]:
-    watch: bool = app.state.watch
-    if watch:
-        session_mgr = app.state.session_manager
+    state = AppState.from_app(app)
+    if state.watch:
+        session_mgr = state.session_manager
         session_mgr.start_file_watcher()
     yield
 
 
 @contextlib.asynccontextmanager
 async def open_browser(app: Starlette) -> AsyncIterator[None]:
-    headless = app.state.headless
-    if not headless:
-        host = app.state.host
-        port = app.state.port
-        base_url = app.state.base_url
-        url = f"http://{host}:{port}{base_url}"
-        user_config = app.state.config_manager.get_config()
+    state = AppState.from_app(app)
+    if not state.headless:
+        url = _startup_url(state)
+        user_config = state.config_manager.get_config()
         browser = user_config["server"]["browser"]
         # Wait 20ms for the server to start and then open the browser, but this
         # function must complete
@@ -110,32 +110,16 @@ async def open_browser(app: Starlette) -> AsyncIterator[None]:
 
 @contextlib.asynccontextmanager
 async def logging(app: Starlette) -> AsyncIterator[None]:
-    manager: SessionManager = app.state.session_manager
-    host = app.state.host
-    port = app.state.port
-    base_url = app.state.base_url
+    state = AppState.from_app(app)
+    manager: SessionManager = state.session_manager
     file_router = manager.file_router
 
-    try:
-        # pretty printing:
-        # if the address maps to localhost, print "localhost" to stdout
-        if (
-            socket.getnameinfo((host, port), socket.NI_NOFQDN)[0]
-            == "localhost"
-        ):
-            host = "localhost"
-    except Exception:
-        # aggressive try/except in case of platform-specific quirks;
-        # nothing to handle, since the `try` logic is just for pretty
-        # printing the host name
-        ...
-
     # Startup message
     if not manager.quiet:
         file = file_router.maybe_get_single_file()
         print_startup(
             file.name if file else None,
-            f"http://{host}:{port}{base_url}",
+            _startup_url(state),
             manager.mode == SessionMode.RUN,
             new=file_router.get_unique_file_key() == AppFileRouter.NEW_FILE,
         )
@@ -149,13 +133,14 @@ async def logging(app: Starlette) -> AsyncIterator[None]:
 
 @contextlib.asynccontextmanager
 async def signal_handler(app: Starlette) -> AsyncIterator[None]:
-    manager = app.state.session_manager
+    state = AppState.from_app(app)
+    manager = state.session_manager
 
     # Interrupt handler
     def shutdown() -> None:
         manager.shutdown()
-        if app.state.server:
-            close_uvicorn(app.state.server)
+        if state.server:
+            close_uvicorn(state.server)
 
     InterruptHandler(
         quiet=manager.quiet,
@@ -170,3 +155,26 @@ async def etc(app: Starlette) -> AsyncIterator[None]:
     # Mimetypes
     initialize_mimetypes()
     yield
+
+
+def _startup_url(state: AppStateBase) -> str:
+    host = state.host
+    port = state.port
+    try:
+        # pretty printing:
+        # if the address maps to localhost, print "localhost" to stdout
+        if (
+            socket.getnameinfo((host, port), socket.NI_NOFQDN)[0]
+            == "localhost"
+        ):
+            host = "localhost"
+    except Exception:
+        # aggressive try/except in case of platform-specific quirks;
+        # nothing to handle, since the `try` logic is just for pretty
+        # printing the host name
+        ...
+
+    url = f"http://{host}:{port}{state.base_url}"
+    if AuthToken.is_empty(state.session_manager.auth_token):
+        return url
+    return f"{url}?access_token={str(state.session_manager.auth_token)}"
diff --git a/marimo/_server/api/middleware.py b/marimo/_server/api/middleware.py
index 18277bba65b..9772f22e1de 100644
--- a/marimo/_server/api/middleware.py
+++ b/marimo/_server/api/middleware.py
@@ -3,6 +3,7 @@
 
 from typing import TYPE_CHECKING, Optional
 
+import starlette.status as status
 from starlette.authentication import (
     AuthCredentials,
     AuthenticationBackend,
@@ -12,27 +13,47 @@
 from starlette.requests import HTTPConnection, Request
 from starlette.responses import JSONResponse
 
+from marimo._server.api.auth import validate_auth
+from marimo._server.api.deps import AppState, AppStateBase
 from marimo._server.model import SessionMode
 
 if TYPE_CHECKING:
+    from starlette.requests import HTTPConnection
     from starlette.types import ASGIApp, Receive, Scope, Send
 
 
 class AuthBackend(AuthenticationBackend):
+    def __init__(self, should_authenticate: bool = True) -> None:
+        self.should_authenticate = should_authenticate
+
     async def authenticate(
         self, conn: HTTPConnection
     ) -> Optional[tuple["AuthCredentials", "BaseUser"]]:
-        mode = conn.app.state.session_manager.mode
-        if mode is None:
-            return None
+        mode = AppStateBase(conn.app.state).session_manager.mode
+
+        # We may not need to authenticate. This can be disabled
+        # because the user is running in a trusted environment
+        # or authentication is handled by a layer above us
+        if self.should_authenticate:
+            # Valid auth header
+            # This validates we have a valid Cookie (already authenticated)
+            # or validates our auth (and sets the cookie)
+            valid = validate_auth(conn)
+            if not valid:
+                return None
+
+        # User's get Read access in Run mode
         if mode == SessionMode.RUN:
             return AuthCredentials(["read"]), SimpleUser("user")
-        elif mode == SessionMode.EDIT:
+
+        # User's get Read and Edit access in Edit mode
+        if mode == SessionMode.EDIT:
             return AuthCredentials(["read", "edit"]), SimpleUser("user")
-        return None
 
+        raise ValueError(f"Invalid session mode: {mode}")
 
-class ValidateServerTokensMiddleware:
+
+class SkewProtectionMiddleware:
     def __init__(self, app: ASGIApp) -> None:
         self.app = app
 
@@ -43,6 +64,7 @@ async def __call__(
             return await self.app(scope, receive, send)
 
         request = Request(scope)
+        state = AppState.from_app(request.app)
 
         # If not POST request, then skip
         if request.method != "POST":
@@ -53,14 +75,12 @@ async def __call__(
         ):
             return await self.app(scope, receive, send)
 
-        expected_server_token = request.app.state.session_manager.server_token
-        if expected_server_token is None:
-            return await self.app(scope, receive, send)
-
+        expected = state.session_manager.skew_protection_token
         server_token = request.headers.get("Marimo-Server-Token")
-        if server_token != expected_server_token:
+        if server_token != str(expected):
             response = JSONResponse(
-                {"error": "Invalid server token"}, status_code=401
+                {"error": "Invalid server token"},
+                status_code=status.HTTP_401_UNAUTHORIZED,
             )
             return await response(scope, receive, send)
 
diff --git a/marimo/_server/asgi.py b/marimo/_server/asgi.py
index 5e7cacc80dc..ab2f2d8a38f 100644
--- a/marimo/_server/asgi.py
+++ b/marimo/_server/asgi.py
@@ -2,9 +2,10 @@
 from __future__ import annotations
 
 import abc
-from typing import TYPE_CHECKING, List, Tuple
+from typing import TYPE_CHECKING, List, Optional, Tuple
 
 from marimo._server.file_router import AppFileRouter
+from marimo._server.tokens import AuthToken
 from marimo._utils.marimo_path import MarimoPath
 
 if TYPE_CHECKING:
@@ -25,6 +26,7 @@ def create_asgi_app(
     *,
     quiet: bool = False,
     include_code: bool = False,
+    token: Optional[str] = None,
 ) -> ASGIAppBuilder:
     """
     Public API to create an ASGI app that can serve multiple notebooks.
@@ -87,6 +89,8 @@ async def root():
 
     - quiet (bool, optional): Suppress standard out
     - include_code (bool, optional): Include notebook code in the app
+    - token (str, optional): Auth token to use for the app.
+        If not provided, an empty token is used.
 
     **Returns.**
 
@@ -105,6 +109,14 @@ async def root():
     user_config_mgr = UserConfigManager()
     base_app = Starlette()
 
+    # Default to an empty token
+    # If a user is using the create_asgi_app API,
+    # they likely want to provide their own authN/authZ
+    if not token:
+        auth_token = AuthToken("")
+    else:
+        auth_token = AuthToken(token)
+
     # We call the entrypoint `root` instead of `filename` incase we want to
     # support directories or code in the future
     class Builder(ASGIAppBuilder):
@@ -130,6 +142,7 @@ def _build_app(self, path: str, root: str) -> "ASGIAppBuilder":
                 # since we don't want to read arbitrary args and apply them
                 # to each application
                 cli_args={},
+                auth_token=auth_token,
             )
             app = create_starlette_app(
                 base_url="",
@@ -140,6 +153,7 @@ def _build_app(self, path: str, root: str) -> "ASGIAppBuilder":
                         lifespans.signal_handler,
                     ]
                 ),
+                enable_auth=not AuthToken.is_empty(auth_token),
             )
             app.state.session_manager = session_manager
             app.state.base_url = path
diff --git a/marimo/_server/export/exporter.py b/marimo/_server/export/exporter.py
index e53f0d5ac01..b100e26d4a1 100644
--- a/marimo/_server/export/exporter.py
+++ b/marimo/_server/export/exporter.py
@@ -25,6 +25,7 @@
 from marimo._server.models.export import ExportAsHTMLRequest
 from marimo._server.session.session_view import SessionView
 from marimo._server.templates.templates import static_notebook_template
+from marimo._server.tokens import SkewProtectionToken
 from marimo._utils.paths import import_files
 
 # Root directory for static assets
@@ -74,7 +75,7 @@ def export_as_html(
         html = static_notebook_template(
             html=index_html,
             user_config=config,
-            server_token="static",
+            server_token=SkewProtectionToken("static"),
             app_config=file_manager.app.config,
             filename=file_manager.filename,
             code=code,
diff --git a/marimo/_server/main.py b/marimo/_server/main.py
index 2ac32642bc9..2c8ccca2f9d 100644
--- a/marimo/_server/main.py
+++ b/marimo/_server/main.py
@@ -11,9 +11,14 @@
 from starlette.responses import JSONResponse
 
 from marimo import _loggers
+from marimo._server.api.auth import (
+    RANDOM_SECRET,
+    CustomSessionMiddleware,
+    on_auth_error,
+)
 from marimo._server.api.middleware import (
     AuthBackend,
-    ValidateServerTokensMiddleware,
+    SkewProtectionMiddleware,
 )
 from marimo._server.api.router import build_routes
 from marimo._server.api.status import (
@@ -32,15 +37,25 @@
 async def handle_error(request: Request, response: Any) -> Any:
     del request
     if isinstance(response, HTTPException):
+        # Turn 403s into 401s to collect auth
+        if response.status_code == 403:
+            return JSONResponse(
+                status_code=401,
+                content={"detail": "Authorization header required"},
+                headers={"WWW-Authenticate": "Basic"},
+            )
         return JSONResponse(
-            {"detail": response.detail}, status_code=response.status_code
+            {"detail": response.detail},
+            status_code=response.status_code,
+            headers=response.headers,
         )
     if isinstance(response, MarimoHTTPException):
         # Log server errors
         if not is_client_error(response.status_code):
             LOGGER.exception(response)
         return JSONResponse(
-            {"detail": response.detail}, status_code=response.status_code
+            {"detail": response.detail},
+            status_code=response.status_code,
         )
     if isinstance(response, TypeError):
         return JSONResponse({"detail": str(response)}, status_code=500)
@@ -51,21 +66,41 @@ async def handle_error(request: Request, response: Any) -> Any:
 
 # Create app
 def create_starlette_app(
+    *,
     base_url: str,
     middleware: Optional[List[Middleware]] = None,
     lifespan: Optional[Lifespan[Starlette]] = None,
+    enable_auth: bool = True,
 ) -> Starlette:
-    final_middlewares: List[Middleware] = [
-        Middleware(
-            CORSMiddleware,
-            allow_origins=["*"],
-            allow_credentials=True,
-            allow_methods=["*"],
-            allow_headers=["*"],
-        ),
-        Middleware(AuthenticationMiddleware, backend=AuthBackend()),
-        Middleware(ValidateServerTokensMiddleware),
-    ]
+    final_middlewares: List[Middleware] = []
+
+    if enable_auth:
+        final_middlewares.extend(
+            [
+                Middleware(
+                    CustomSessionMiddleware,
+                    secret_key=RANDOM_SECRET,
+                ),
+            ]
+        )
+
+    final_middlewares.extend(
+        [
+            Middleware(
+                AuthenticationMiddleware,
+                backend=AuthBackend(should_authenticate=enable_auth),
+                on_error=on_auth_error,
+            ),
+            Middleware(
+                CORSMiddleware,
+                allow_origins=["*"],
+                allow_credentials=True,
+                allow_methods=["*"],
+                allow_headers=["*"],
+            ),
+            Middleware(SkewProtectionMiddleware),
+        ]
+    )
 
     if middleware:
         final_middlewares.extend(middleware)
diff --git a/marimo/_server/sessions.py b/marimo/_server/sessions.py
index 84765a15004..2468eda2000 100644
--- a/marimo/_server/sessions.py
+++ b/marimo/_server/sessions.py
@@ -26,7 +26,6 @@
 from multiprocessing.queues import Queue as MPQueue
 from pathlib import Path
 from typing import Any, Optional
-from uuid import uuid4
 
 from marimo import _loggers
 from marimo._ast.cell import CellConfig, CellId_t
@@ -58,6 +57,7 @@
 from marimo._server.models.models import InstantiateRequest
 from marimo._server.recents import RecentFilesManager
 from marimo._server.session.session_view import SessionView
+from marimo._server.tokens import AuthToken, SkewProtectionToken
 from marimo._server.types import QueueType
 from marimo._server.utils import print_tabbed
 from marimo._utils.disposable import Disposable
@@ -438,7 +438,8 @@ class SessionManager:
     The SessionManager also encapsulates state common to all sessions:
     - the app filename
     - the app mode (edit or run)
-    - the server token
+    - the auth token
+    - the skew-protection token
     """
 
     def __init__(
@@ -451,6 +452,7 @@ def __init__(
         lsp_server: LspServer,
         user_config_manager: UserConfigManager,
         cli_args: SerializedCLIArgs,
+        auth_token: Optional[AuthToken],
     ) -> None:
         self.file_router = file_router
         self.mode = mode
@@ -464,18 +466,23 @@ def __init__(
         self.user_config_manager = user_config_manager
         self.cli_args = cli_args
 
-        if mode == SessionMode.EDIT:
-            # In edit mode, the server gets a random token to prevent
-            # frontends that it didn't create from connecting to it and
-            # executing edit-only commands (such as overwriting the file).
-            self.server_token = str(uuid4())
+        # Auth token and Skew-protection token
+        if auth_token is not None:
+            self.auth_token = auth_token
+            self.skew_protection_token = SkewProtectionToken.random()
+        elif mode == SessionMode.EDIT:
+            # In edit mode, if no auth token is provided,
+            # generate a random token
+            self.auth_token = AuthToken.random()
+            self.skew_protection_token = SkewProtectionToken.random()
         else:
-            # Because run-mode is read-only, all that matters is that
-            # the frontend's app matches the server's app.
             app = file_router.get_single_app_file_manager().app
-            self.server_token = str(
-                hash("".join(code for code in app.cell_manager.codes()))
-            )
+            codes = "".join(code for code in app.cell_manager.codes())
+            # Because run-mode is read-only and we could have multiple
+            # servers for the same app (going to sleep or autoscaling),
+            # we default to a token based on the app's code
+            self.auth_token = AuthToken.from_code(codes)
+            self.skew_protection_token = SkewProtectionToken.from_code(codes)
 
     def app_manager(self, key: MarimoFileKey) -> AppFileManager:
         """
diff --git a/marimo/_server/start.py b/marimo/_server/start.py
index 3a168b13825..c0a19ba58a1 100644
--- a/marimo/_server/start.py
+++ b/marimo/_server/start.py
@@ -13,6 +13,7 @@
 from marimo._server.main import create_starlette_app
 from marimo._server.model import SessionMode
 from marimo._server.sessions import LspServer, SessionManager
+from marimo._server.tokens import AuthToken
 from marimo._server.utils import find_free_port, initialize_asyncio
 from marimo._server.uvicorn_utils import initialize_signals
 from marimo._utils.paths import import_files
@@ -33,6 +34,7 @@ def start(
     watch: bool,
     cli_args: SerializedCLIArgs,
     base_url: str = "",
+    auth_token: Optional[AuthToken],
 ) -> None:
     """
     Start the server.
@@ -52,6 +54,7 @@ def start(
         lsp_server=LspServer(port * 10),
         user_config_manager=user_config_mgr,
         cli_args=cli_args,
+        auth_token=auth_token,
     )
 
     log_level = "info" if development_mode else "error"
@@ -68,6 +71,7 @@ def start(
                 lifespans.open_browser,
             ]
         ),
+        enable_auth=not AuthToken.is_empty(session_manager.auth_token),
     )
 
     app.state.headless = headless
diff --git a/marimo/_server/templates/templates.py b/marimo/_server/templates/templates.py
index 42ffb40c1bf..ac8d45d2cbc 100644
--- a/marimo/_server/templates/templates.py
+++ b/marimo/_server/templates/templates.py
@@ -13,17 +13,18 @@
 from marimo._output.utils import uri_encode_component
 from marimo._server.api.utils import parse_title
 from marimo._server.model import SessionMode
+from marimo._server.tokens import SkewProtectionToken
 
 
 def home_page_template(
     html: str,
     base_url: str,
     user_config: MarimoConfig,
-    server_token: str,
+    server_token: SkewProtectionToken,
 ) -> str:
     html = html.replace("{{ base_url }}", base_url)
     html = html.replace("{{ user_config }}", json.dumps(user_config))
-    html = html.replace("{{ server_token }}", server_token)
+    html = html.replace("{{ server_token }}", str(server_token))
     html = html.replace("{{ version }}", __version__)
 
     html = html.replace("{{ title }}", "marimo")
@@ -38,14 +39,14 @@ def notebook_page_template(
     html: str,
     base_url: str,
     user_config: MarimoConfig,
-    server_token: str,
+    server_token: SkewProtectionToken,
     app_config: _AppConfig,
     filename: Optional[str],
     mode: SessionMode,
 ) -> str:
     html = html.replace("{{ base_url }}", base_url)
     html = html.replace("{{ user_config }}", json.dumps(user_config))
-    html = html.replace("{{ server_token }}", server_token)
+    html = html.replace("{{ server_token }}", str(server_token))
     html = html.replace("{{ version }}", __version__)
 
     html = html.replace(
@@ -67,7 +68,7 @@ def notebook_page_template(
 def static_notebook_template(
     html: str,
     user_config: MarimoConfig,
-    server_token: str,
+    server_token: SkewProtectionToken,
     app_config: _AppConfig,
     filename: Optional[str],
     code: str,
@@ -87,7 +88,7 @@ def static_notebook_template(
     html = html.replace(
         "{{ user_config }}", json.dumps(user_config, sort_keys=True)
     )
-    html = html.replace("{{ server_token }}", server_token)
+    html = html.replace("{{ server_token }}", str(server_token))
     html = html.replace("{{ version }}", __version__)
 
     html = html.replace(
diff --git a/marimo/_server/tokens.py b/marimo/_server/tokens.py
new file mode 100644
index 00000000000..acb1bf825a3
--- /dev/null
+++ b/marimo/_server/tokens.py
@@ -0,0 +1,44 @@
+# Copyright 2024 Marimo. All rights reserved.
+from __future__ import annotations
+
+import secrets
+
+from starlette.datastructures import Secret
+
+
+class AuthToken(Secret):
+    @staticmethod
+    def random() -> AuthToken:
+        return AuthToken(secrets.token_urlsafe(16))
+
+    @staticmethod
+    def from_code(code: str) -> AuthToken:
+        return AuthToken(str(hash(code)))
+
+    @staticmethod
+    def is_empty(token: AuthToken) -> bool:
+        return str(token) == ""
+
+
+class SkewProtectionToken:
+    """
+    Provides a token that is sent to the client on the first request and
+    is used to protect against version skew bugs.
+
+    This can happen when new code is deployed to the server but the client
+    still has only application loaded.
+    """
+
+    def __init__(self, token: str) -> None:
+        self.token = token
+
+    @staticmethod
+    def from_code(code: str) -> SkewProtectionToken:
+        return SkewProtectionToken(str(hash(code)))
+
+    @staticmethod
+    def random() -> SkewProtectionToken:
+        return SkewProtectionToken(secrets.token_urlsafe(16))
+
+    def __str__(self) -> str:
+        return self.token
diff --git a/pyproject.toml b/pyproject.toml
index 2f79c8734df..586dc3b6dfd 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -39,6 +39,8 @@ dependencies = [
     "docutils>=0.17.0",
     # to show RAM, CPU usage, other system utilities
     "psutil>=5.0",
+    # required dependency in Starlette for SessionMiddleware support
+    "itsdangerous>=2.0.0",
     # for cell formatting; if user version is not compatible, no-op
     # so no lower bound needed
     "black",
diff --git a/tests/_cli/test_cli.py b/tests/_cli/test_cli.py
index cd9b6199514..bdddd89a431 100644
--- a/tests/_cli/test_cli.py
+++ b/tests/_cli/test_cli.py
@@ -57,7 +57,7 @@ def _confirm_shutdown(process: subprocess.Popen) -> None:
 
 
 def _check_shutdown(
-    process, check_fn: Optional[Callable[[int], bool]] = None
+    process: subprocess.Popen, check_fn: Optional[Callable[[int], bool]] = None
 ) -> None:
     max_tries = 3
     tries = 0
@@ -70,11 +70,16 @@ def _check_shutdown(
         assert check_fn(process.poll())
 
 
-def _try_fetch(port: int, host: str = "localhost") -> Optional[bytes]:
+def _try_fetch(
+    port: int, host: str = "localhost", token: Optional[str] = None
+) -> Optional[bytes]:
     contents = None
     for _ in range(10):
         try:
-            contents = urllib.request.urlopen(f"http://{host}:{port}").read()
+            url = f"http://{host}:{port}"
+            if token is not None:
+                url = f"{url}?access_token={token}"
+            contents = urllib.request.urlopen(url).read()
             break
         except Exception:
             time.sleep(0.5)
@@ -135,7 +140,9 @@ def test_cli_edit_none() -> None:
     # Python 3.8 compatibility, such as forgetting `from __future__` import
     # annotations
     port = _get_port()
-    p = subprocess.Popen(["marimo", "edit", "-p", str(port), "--headless"])
+    p = subprocess.Popen(
+        ["marimo", "edit", "-p", str(port), "--headless", "--no-token"]
+    )
     contents = _try_fetch(port)
     _check_contents(p, b"marimo-mode data-mode='home'", contents)
     _check_contents(
@@ -144,11 +151,36 @@ def test_cli_edit_none() -> None:
     _check_contents(p, b"marimo-server-token", contents)
 
 
+def test_cli_edit_token() -> None:
+    # smoke test: makes sure CLI starts and has basic things we expect
+    # helpful for catching issues related to
+    # Python 3.8 compatibility, such as forgetting `from __future__` import
+    # annotations
+    port = _get_port()
+    p = subprocess.Popen(
+        [
+            "marimo",
+            "edit",
+            "-p",
+            str(port),
+            "--headless",
+            "--token-password",
+            "secret",
+        ]
+    )
+    contents = _try_fetch(port, "localhost", "secret")
+    _check_contents(p, b"marimo-mode data-mode='home'", contents)
+    _check_contents(
+        p, f"marimo-version data-version='{__version__}'".encode(), contents
+    )
+    _check_contents(p, b"marimo-server-token", contents)
+
+
 def test_cli_edit_directory() -> None:
     d = tempfile.TemporaryDirectory()
     port = _get_port()
     p = subprocess.Popen(
-        ["marimo", "edit", d.name, "-p", str(port), "--headless"]
+        ["marimo", "edit", d.name, "-p", str(port), "--headless", "--no-token"]
     )
     contents = _try_fetch(port)
     _check_contents(p, b"marimo-mode data-mode='home'", contents)
@@ -163,7 +195,7 @@ def test_cli_edit_new_file() -> None:
     path = os.path.join(d.name, "new.py")
     port = _get_port()
     p = subprocess.Popen(
-        ["marimo", "edit", path, "-p", str(port), "--headless"]
+        ["marimo", "edit", path, "-p", str(port), "--headless", "--no-token"]
     )
     contents = _try_fetch(port)
     _check_contents(p, b"marimo-mode data-mode='edit'", contents)
@@ -183,6 +215,7 @@ def test_cli_edit_with_additional_args(temp_marimo_file: str) -> None:
             "-p",
             str(port),
             "--headless",
+            "--no-token",
             "--",
             "-a=foo",
             "--b=bar",
@@ -197,7 +230,9 @@ def test_cli_edit_with_additional_args(temp_marimo_file: str) -> None:
 
 def test_cli_new() -> None:
     port = _get_port()
-    p = subprocess.Popen(["marimo", "new", "-p", str(port), "--headless"])
+    p = subprocess.Popen(
+        ["marimo", "new", "-p", str(port), "--headless", "--no-token"]
+    )
     contents = _try_fetch(port)
     _check_contents(p, b"marimo-mode data-mode='edit'", contents)
     _check_contents(
@@ -263,7 +298,15 @@ def test_cli_run_with_additional_args(temp_marimo_file: str) -> None:
 def test_cli_tutorial() -> None:
     port = _get_port()
     p = subprocess.Popen(
-        ["marimo", "tutorial", "intro", "-p", str(port), "--headless"]
+        [
+            "marimo",
+            "tutorial",
+            "intro",
+            "-p",
+            str(port),
+            "--headless",
+            "--no-token",
+        ]
     )
     contents = _try_fetch(port)
     _check_contents(p, b"marimo-mode data-mode='edit'", contents)
@@ -284,6 +327,7 @@ def test_cli_custom_host() -> None:
             "-p",
             str(port),
             "--headless",
+            "--no-token",
             "--host",
             host,
         ]
@@ -296,7 +340,9 @@ def test_cli_custom_host() -> None:
 def test_cli_edit_interrupt_twice() -> None:
     # two SIGINTs should kill the CLI
     port = _get_port()
-    p = subprocess.Popen(["marimo", "edit", "-p", str(port), "--headless"])
+    p = subprocess.Popen(
+        ["marimo", "edit", "-p", str(port), "--headless", "--no-token"]
+    )
     _check_started(port)
     with _patch_signals_win32():
         _interrupt(p)
@@ -328,7 +374,7 @@ def test_cli_run_interrupt_twice() -> None:
 def test_cli_edit_shutdown() -> None:
     port = _get_port()
     p = subprocess.Popen(
-        ["marimo", "edit", "-p", str(port), "--headless"],
+        ["marimo", "edit", "-p", str(port), "--headless", "--no-token"],
         stdin=subprocess.PIPE,
     )
     _check_started(port)
@@ -346,7 +392,7 @@ def test_cli_run_shutdown() -> None:
     path = _temp_run_file(d)
     port = _get_port()
     p = subprocess.Popen(
-        ["marimo", "run", path, "-p", str(port), "--headless"],
+        ["marimo", "run", path, "-p", str(port), "--headless", "--no-token"],
         stdin=subprocess.PIPE,
     )
     _check_started(port)
diff --git a/tests/_server/api/endpoints/test_ai.py b/tests/_server/api/endpoints/test_ai.py
index 84dfe45c077..038998182c7 100644
--- a/tests/_server/api/endpoints/test_ai.py
+++ b/tests/_server/api/endpoints/test_ai.py
@@ -1,21 +1,25 @@
 # Copyright 2024 Marimo. All rights reserved.
+from __future__ import annotations
+
 from contextlib import contextmanager
 from dataclasses import dataclass
-from typing import Any, List
+from typing import TYPE_CHECKING, Any, List
 from unittest.mock import MagicMock, patch
 
 import pytest
-from starlette.testclient import TestClient
 
 from marimo._config.manager import UserConfigManager
 from marimo._dependencies.dependencies import DependencyManager
 from tests._server.conftest import get_user_config_manager
-from tests._server.mocks import with_session
+from tests._server.mocks import token_header, with_session
+
+if TYPE_CHECKING:
+    from starlette.testclient import TestClient
 
 SESSION_ID = "session-123"
 HEADERS = {
     "Marimo-Session-Id": SESSION_ID,
-    "Marimo-Server-Token": "fake-token",
+    **token_header("fake-token"),
 }
 
 HAS_DEPS = DependencyManager.has_openai()
diff --git a/tests/_server/api/endpoints/test_assets.py b/tests/_server/api/endpoints/test_assets.py
index 15b4a1e4d69..3f8c6233e88 100644
--- a/tests/_server/api/endpoints/test_assets.py
+++ b/tests/_server/api/endpoints/test_assets.py
@@ -1,20 +1,26 @@
 # Copyright 2024 Marimo. All rights reserved.
-from tempfile import TemporaryDirectory
-from typing import Any, cast
+from __future__ import annotations
 
-from starlette.testclient import TestClient
+from tempfile import TemporaryDirectory
+from typing import TYPE_CHECKING, Any, cast
 
 from marimo._server.api.utils import parse_title
 from marimo._server.file_router import AppFileRouter
 from marimo._server.sessions import SessionManager
-from tests._server.mocks import with_file_router
+from tests._server.mocks import token_header, with_file_router
+
+if TYPE_CHECKING:
+    from starlette.testclient import TestClient
 
 
 def test_index(client: TestClient) -> None:
     session_manager: SessionManager = cast(
         Any, client.app
     ).state.session_manager
-    response = client.get("/")
+    response = client.get("/")  # no header
+    assert response.status_code == 401, response.text
+
+    response = client.get("/", headers=token_header())
     assert response.status_code == 200, response.text
     content = response.text
     filename = session_manager.file_router.get_unique_file_key()
@@ -24,35 +30,32 @@ def test_index(client: TestClient) -> None:
     assert filename in content
     assert "<marimo-mode data-mode='edit'" in content
     assert f"<title>{title}</title>" in content
-    assert session_manager.server_token in content
 
 
 @with_file_router(AppFileRouter.from_files([]))
 def test_index_when_empty(client: TestClient) -> None:
-    session_manager: SessionManager = cast(
-        Any, client.app
-    ).state.session_manager
-    response = client.get("/")
+    response = client.get("/")  # no header
+    assert response.status_code == 401, response.text
+
+    response = client.get("/", headers=token_header())
     assert response.status_code == 200, response.text
     content = response.text
     assert "<marimo-filename hidden></marimo-filename>" in content
     assert "<marimo-mode data-mode='home'" in content
     assert "<title>marimo</title>" in content
-    assert session_manager.server_token in content
 
 
 @with_file_router(AppFileRouter.new_file())
 def test_index_when_new_file(client: TestClient) -> None:
-    session_manager: SessionManager = cast(
-        Any, client.app
-    ).state.session_manager
-    response = client.get("/")
+    response = client.get("/")  # no header
+    assert response.status_code == 401, response.text
+
+    response = client.get("/", headers=token_header())
     assert response.status_code == 200, response.text
     content = response.text
     assert "<marimo-filename hidden></marimo-filename>" in content
     assert "<marimo-mode data-mode='edit'" in content
     assert "<title>marimo</title>" in content
-    assert session_manager.server_token in content
 
 
 TEMP_DIR = TemporaryDirectory()
@@ -60,16 +63,12 @@ def test_index_when_new_file(client: TestClient) -> None:
 
 @with_file_router(AppFileRouter.from_directory(TEMP_DIR.name))
 def test_index_with_directory(client: TestClient) -> None:
-    session_manager: SessionManager = cast(
-        Any, client.app
-    ).state.session_manager
-    response = client.get("/")
+    response = client.get("/", headers=token_header())
     assert response.status_code == 200, response.text
     content = response.text
     assert "<marimo-filename" in content
     assert "<marimo-mode data-mode='home'" in content
     assert "<title>marimo</title>" in content
-    assert session_manager.server_token in content
 
 
 def test_favicon(client: TestClient) -> None:
@@ -90,12 +89,15 @@ def test_unknown_file(client: TestClient) -> None:
 
 
 def test_vfile(client: TestClient) -> None:
-    response = client.get("/@file/empty.txt")
+    response = client.get("/@file/example.txt")
+    assert response.status_code == 401, response.text
+
+    response = client.get("/@file/empty.txt", headers=token_header())
     assert response.status_code == 200, response.text
     assert response.headers["content-type"] == "application/octet-stream"
     assert response.content == b""
 
-    response = client.get("/@file/bad.txt")
+    response = client.get("/@file/bad.txt", headers=token_header())
     assert response.status_code == 404, response.text
     assert response.headers["content-type"] == "application/json"
     assert response.json() == {"detail": "Invalid virtual file request"}
diff --git a/tests/_server/api/endpoints/test_config_endpoints.py b/tests/_server/api/endpoints/test_config_endpoints.py
index 366727efbcf..befb09f919f 100644
--- a/tests/_server/api/endpoints/test_config_endpoints.py
+++ b/tests/_server/api/endpoints/test_config_endpoints.py
@@ -4,7 +4,7 @@
 from typing import TYPE_CHECKING
 
 from tests._server.conftest import get_user_config_manager
-from tests._server.mocks import with_session
+from tests._server.mocks import token_header, with_session
 
 if TYPE_CHECKING:
     from starlette.testclient import TestClient
@@ -12,7 +12,7 @@
 SESSION_ID = "session-123"
 HEADERS = {
     "Marimo-Session-Id": SESSION_ID,
-    "Marimo-Server-Token": "fake-token",
+    **token_header("fake-token"),
 }
 
 
diff --git a/tests/_server/api/endpoints/test_editing.py b/tests/_server/api/endpoints/test_editing.py
index 7f2b81fff32..7ad5366eb17 100644
--- a/tests/_server/api/endpoints/test_editing.py
+++ b/tests/_server/api/endpoints/test_editing.py
@@ -1,14 +1,17 @@
 # Copyright 2024 Marimo. All rights reserved.
+from __future__ import annotations
 
+from typing import TYPE_CHECKING
 
-from starlette.testclient import TestClient
+from tests._server.mocks import token_header, with_session
 
-from tests._server.mocks import with_session
+if TYPE_CHECKING:
+    from starlette.testclient import TestClient
 
 SESSION_ID = "session-123"
 HEADERS = {
     "Marimo-Session-Id": SESSION_ID,
-    "Marimo-Server-Token": "fake-token",
+    **token_header("fake-token"),
 }
 
 
diff --git a/tests/_server/api/endpoints/test_execution.py b/tests/_server/api/endpoints/test_execution.py
index 3ebc3132b87..b6d7c356b6f 100644
--- a/tests/_server/api/endpoints/test_execution.py
+++ b/tests/_server/api/endpoints/test_execution.py
@@ -1,15 +1,18 @@
 # Copyright 2024 Marimo. All rights reserved.
+from __future__ import annotations
 
-
-from starlette.testclient import TestClient
+from typing import TYPE_CHECKING
 
 from tests._server.conftest import get_session_manager
-from tests._server.mocks import with_read_session, with_session
+from tests._server.mocks import token_header, with_read_session, with_session
+
+if TYPE_CHECKING:
+    from starlette.testclient import TestClient
 
 SESSION_ID = "session-123"
 HEADERS = {
     "Marimo-Session-Id": SESSION_ID,
-    "Marimo-Server-Token": "fake-token",
+    **token_header("fake-token"),
 }
 
 
@@ -80,10 +83,10 @@ def test_restart_session(client: TestClient) -> None:
         assert response.status_code == 200, response.text
         assert response.headers["content-type"] == "application/json"
         assert "success" in response.json()
-        server_token: str = get_session_manager(client).server_token
+        auth_token = get_session_manager(client).auth_token
         client.post(
             "/api/kernel/shutdown",
-            headers={"Marimo-Server-Token": server_token},
+            headers=token_header(auth_token),
         )
 
     @staticmethod
@@ -154,13 +157,13 @@ def test_function_call(client: TestClient) -> None:
     @with_read_session(SESSION_ID)
     def test_interrupt(client: TestClient) -> None:
         response = client.post("/api/kernel/interrupt", headers=HEADERS)
-        assert response.status_code == 403, response.text
+        assert response.status_code == 401, response.text
 
     @staticmethod
     @with_read_session(SESSION_ID)
     def test_restart_session(client: TestClient) -> None:
         response = client.post("/api/kernel/restart_session", headers=HEADERS)
-        assert response.status_code == 403, response.text
+        assert response.status_code == 401, response.text
 
     @staticmethod
     @with_read_session(SESSION_ID)
@@ -173,4 +176,4 @@ def test_run_cell(client: TestClient) -> None:
                 "codes": ["print('Hello, cell-1')", "print('Hello, cell-2')"],
             },
         )
-        assert response.status_code == 403, response.text
+        assert response.status_code == 401, response.text
diff --git a/tests/_server/api/endpoints/test_export.py b/tests/_server/api/endpoints/test_export.py
index 8246d4e4d0a..154968bf3af 100644
--- a/tests/_server/api/endpoints/test_export.py
+++ b/tests/_server/api/endpoints/test_export.py
@@ -5,7 +5,7 @@
 
 from marimo import __version__
 from tests._server.conftest import get_session_manager
-from tests._server.mocks import with_session
+from tests._server.mocks import token_header, with_session
 from tests.mocks import snapshotter
 
 if TYPE_CHECKING:
@@ -16,7 +16,7 @@
 SESSION_ID = "session-123"
 HEADERS = {
     "Marimo-Session-Id": SESSION_ID,
-    "Marimo-Server-Token": "fake-token",
+    **token_header("fake-token"),
 }
 
 
@@ -38,6 +38,27 @@ def test_export_html(client: TestClient) -> None:
     assert '<marimo-code hidden=""></marimo-code>' not in body
 
 
+@with_session(SESSION_ID)
+def test_export_html_skew_protection(client: TestClient) -> None:
+    session = get_session_manager(client).get_session(SESSION_ID)
+    assert session
+    session.app_file_manager.filename = "test.py"
+    response = client.post(
+        "/api/export/html",
+        headers={
+            **HEADERS,
+            "Marimo-Server-Token": "old-skew-id",
+        },
+        json={
+            "download": False,
+            "files": [],
+            "include_code": True,
+        },
+    )
+    assert response.status_code == 401
+    assert response.json() == {"error": "Invalid server token"}
+
+
 @with_session(SESSION_ID)
 def test_export_html_no_code(client: TestClient) -> None:
     session = get_session_manager(client).get_session(SESSION_ID)
diff --git a/tests/_server/api/endpoints/test_file_explorer.py b/tests/_server/api/endpoints/test_file_explorer.py
index 423f3065ff9..3468696cb8d 100644
--- a/tests/_server/api/endpoints/test_file_explorer.py
+++ b/tests/_server/api/endpoints/test_file_explorer.py
@@ -1,13 +1,17 @@
 # Copyright 2024 Marimo. All rights reserved.
-
+from __future__ import annotations
 
 import os
 from tempfile import TemporaryDirectory
+from typing import TYPE_CHECKING
+
+from tests._server.mocks import token_header
 
-from starlette.testclient import TestClient
+if TYPE_CHECKING:
+    from starlette.testclient import TestClient
 
 HEADERS = {
-    "Marimo-Server-Token": "fake-token",
+    **token_header("fake-token"),
 }
 
 temp_dir = TemporaryDirectory()
diff --git a/tests/_server/api/endpoints/test_files.py b/tests/_server/api/endpoints/test_files.py
index 66bb18d3fa4..6319344e912 100644
--- a/tests/_server/api/endpoints/test_files.py
+++ b/tests/_server/api/endpoints/test_files.py
@@ -6,7 +6,7 @@
 from typing import TYPE_CHECKING
 
 from tests._server.conftest import get_session_manager
-from tests._server.mocks import with_session
+from tests._server.mocks import token_header, with_session
 
 if TYPE_CHECKING:
     from fastapi.testclient import TestClient
@@ -14,7 +14,7 @@
 SESSION_ID = "session-123"
 HEADERS = {
     "Marimo-Session-Id": SESSION_ID,
-    "Marimo-Server-Token": "fake-token",
+    **token_header("fake-token"),
 }
 
 
diff --git a/tests/_server/api/endpoints/test_health.py b/tests/_server/api/endpoints/test_health.py
index a506266e232..e5cc8d14ad2 100644
--- a/tests/_server/api/endpoints/test_health.py
+++ b/tests/_server/api/endpoints/test_health.py
@@ -4,6 +4,7 @@
 from typing import TYPE_CHECKING
 
 from marimo import __version__
+from tests._server.mocks import token_header
 
 if TYPE_CHECKING:
     from starlette.testclient import TestClient
@@ -19,7 +20,11 @@ def test_health(client: TestClient) -> None:
 
 
 def test_status(client: TestClient) -> None:
+    # Unauthorized
     response = client.get("/api/status")
+    assert response.status_code == 401, response.text
+
+    response = client.get("/api/status", headers=token_header())
     assert response.status_code == 200, response.text
     content = response.json()
     assert content["status"] == "healthy"
@@ -37,7 +42,11 @@ def test_version(client: TestClient) -> None:
 
 
 def test_memory(client: TestClient) -> None:
-    response = client.get("/api/usage")
+    # Unauthorized
+    response = client.get("/api/status")
+    assert response.status_code == 401, response.text
+
+    response = client.get("/api/usage", headers=token_header())
     assert response.status_code == 200, response.text
     memory = response.json()["memory"]
     assert memory["total"] > 0
diff --git a/tests/_server/api/endpoints/test_home.py b/tests/_server/api/endpoints/test_home.py
index 534905e32e8..e6ed63fed1c 100644
--- a/tests/_server/api/endpoints/test_home.py
+++ b/tests/_server/api/endpoints/test_home.py
@@ -1,17 +1,21 @@
 # Copyright 2024 Marimo. All rights reserved.
+from __future__ import annotations
 
 import sys
+from typing import TYPE_CHECKING
 
 import pytest
-from starlette.testclient import TestClient
 
 from tests._server.conftest import get_session_manager
-from tests._server.mocks import with_session
+from tests._server.mocks import token_header, with_session
+
+if TYPE_CHECKING:
+    from starlette.testclient import TestClient
 
 SESSION_ID = "session-123"
 HEADERS = {
     "Marimo-Session-Id": SESSION_ID,
-    "Marimo-Server-Token": "fake-token",
+    **token_header("fake-token"),
 }
 
 
diff --git a/tests/_server/api/endpoints/test_resume_session.py b/tests/_server/api/endpoints/test_resume_session.py
index 43c0f6a8f9d..44e4935d73c 100644
--- a/tests/_server/api/endpoints/test_resume_session.py
+++ b/tests/_server/api/endpoints/test_resume_session.py
@@ -7,6 +7,7 @@
 from marimo._server.sessions import Session
 from marimo._utils.parse_dataclass import parse_raw
 from tests._server.conftest import get_session_manager
+from tests._server.mocks import token_header
 
 if TYPE_CHECKING:
     from starlette.testclient import TestClient
@@ -33,12 +34,12 @@ def create_response(
 def headers(session_id: str) -> dict[str, str]:
     return {
         "Marimo-Session-Id": session_id,
-        "Marimo-Server-Token": "fake-token",
+        **token_header("fake-token"),
     }
 
 
 HEADERS = {
-    "Marimo-Server-Token": "fake-token",
+    **token_header("fake-token"),
 }
 
 
diff --git a/tests/_server/api/endpoints/test_ws.py b/tests/_server/api/endpoints/test_ws.py
index c4a798fa90b..ba61d667f61 100644
--- a/tests/_server/api/endpoints/test_ws.py
+++ b/tests/_server/api/endpoints/test_ws.py
@@ -12,6 +12,7 @@
 from marimo._server.sessions import SessionManager
 from marimo._utils.parse_dataclass import parse_raw
 from tests._server.conftest import get_session_manager
+from tests._server.mocks import token_header
 
 if TYPE_CHECKING:
     from starlette.testclient import TestClient
@@ -36,7 +37,7 @@ def create_response(
 
 
 HEADERS = {
-    "Marimo-Server-Token": "fake-token",
+    **token_header("fake-token"),
 }
 
 
diff --git a/tests/_server/api/test_auth.py b/tests/_server/api/test_auth.py
new file mode 100644
index 00000000000..93e0fe7dbdb
--- /dev/null
+++ b/tests/_server/api/test_auth.py
@@ -0,0 +1,140 @@
+from __future__ import annotations
+
+import base64
+from typing import Any
+
+import pytest
+from starlette.applications import Starlette
+from starlette.datastructures import Headers, QueryParams
+from starlette.requests import HTTPConnection
+
+from marimo._config.manager import UserConfigManager
+from marimo._server.api.auth import (
+    AppState,
+    CustomSessionMiddleware,
+    validate_auth,
+)
+from marimo._server.main import create_starlette_app
+from tests._server.mocks import get_mock_session_manager
+
+
+async def mock_receive() -> Any:
+    return {}
+
+
+async def mock_send(message: Any) -> None:
+    del message
+    pass
+
+
+async def test_custom_session_middleware_call(app: Starlette):
+    middleware = CustomSessionMiddleware(app, "secret_key")
+    scope = create_connection(app).scope
+
+    await middleware(scope, mock_receive, mock_send)
+    assert middleware.session_cookie == "session_1234"
+
+
+async def test_custom_session_middleware_call_with_port():
+    app = Starlette()
+    middleware = CustomSessionMiddleware(app, "secret_key")
+    scope = create_connection(app).scope
+
+    await middleware(scope, mock_receive, mock_send)
+    assert middleware.session_cookie == "session"
+
+
+@pytest.fixture
+def app() -> Starlette:
+    app = create_starlette_app(base_url="", enable_auth=True)
+    app.state.session_manager = get_mock_session_manager()
+    app.state.config_manager = UserConfigManager()
+    app.state.host = "localhost"
+    app.state.port = 1234
+    app.state.base_url = ""
+    return app
+
+
+def create_connection(app: Starlette) -> HTTPConnection:
+    conn = HTTPConnection(
+        {
+            "type": "http",
+            "app": app,
+            "headers": {},
+            "query_string": "",
+            "method": "GET",
+            "path": "/",
+        }
+    )
+    return conn
+
+
+async def test_validate_auth_with_valid_cookie(app: Starlette):
+    conn = create_connection(app)
+    # Run all middleware
+    await app.build_middleware_stack()(conn.scope, mock_receive, mock_send)
+    conn.session["access_token"] = str(
+        AppState.from_app(app).session_manager.auth_token
+    )
+
+    assert validate_auth(conn) is True
+
+
+async def test_validate_auth_with_bad_cookie(app: Starlette):
+    conn = create_connection(app)
+    # Run all middleware
+    await app.build_middleware_stack()(conn.scope, mock_receive, mock_send)
+    conn.session["access_token"] = "bad_token"
+
+    assert validate_auth(conn) is False
+
+
+async def test_validate_auth_with_valid_access_token(app: Starlette):
+    conn = create_connection(app)
+    # Run all middleware
+    await app.build_middleware_stack()(conn.scope, mock_receive, mock_send)
+    token = str(AppState.from_app(app).session_manager.auth_token)
+    conn._query_params = QueryParams([("access_token", token)])
+
+    assert validate_auth(conn) is True
+
+
+async def test_validate_auth_with_invalid_access_token(app: Starlette):
+    conn = create_connection(app)
+    # Run all middleware
+    await app.build_middleware_stack()(conn.scope, mock_receive, mock_send)
+    conn._query_params = QueryParams([("access_token", "bad_token")])
+
+    assert validate_auth(conn) is False
+
+
+async def test_validate_auth_with_valid_basic_auth(app: Starlette):
+    conn = create_connection(app)
+    auth_token = AppState.from_app(app).session_manager.auth_token
+    # Run all middleware
+    await app.build_middleware_stack()(conn.scope, mock_receive, mock_send)
+    basic_auth_header = (
+        f"Basic {base64.b64encode(f'user:{auth_token}'.encode()).decode()}"
+    )
+    conn._headers = Headers({"Authorization": basic_auth_header})
+
+    assert validate_auth(conn) is True
+
+
+async def test_validate_auth_with_missing_password_in_basic_auth(
+    app: Starlette,
+):
+    conn = create_connection(app)
+    # Run all middleware
+    await app.build_middleware_stack()(conn.scope, mock_receive, mock_send)
+    basic_auth_header = f"Basic {base64.b64encode('test:'.encode()).decode()}"
+    conn._headers = Headers({"Authorization": basic_auth_header})
+
+    assert validate_auth(conn) is False
+
+
+async def test_validate_auth_with_no_auth(app: Starlette):
+    conn = create_connection(app)
+    # Run all middleware
+    await app.build_middleware_stack()(conn.scope, mock_receive, mock_send)
+    assert validate_auth(conn) is False
diff --git a/tests/_server/api/test_middleware.py b/tests/_server/api/test_middleware.py
index 13a3c09ebb7..744f1cb2864 100644
--- a/tests/_server/api/test_middleware.py
+++ b/tests/_server/api/test_middleware.py
@@ -1,10 +1,20 @@
 # Copyright 2024 Marimo. All rights reserved.
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Any
+
+import pytest
 import uvicorn
 from starlette.testclient import TestClient
 
 from marimo._config.manager import UserConfigManager
 from marimo._server.main import create_starlette_app
-from tests._server.mocks import get_mock_session_manager
+from marimo._server.model import SessionMode
+from marimo._server.tokens import AuthToken
+from tests._server.mocks import get_mock_session_manager, token_header
+
+if TYPE_CHECKING:
+    from starlette.applications import Starlette
 
 
 def test_base_url() -> None:
@@ -34,8 +44,8 @@ def test_base_url() -> None:
     response = client.get("/healthz")
     assert response.status_code == 404, response.text
 
-    # Index
-    response = client.get("/foo/")
+    # Index (requires auth)
+    response = client.get("/foo/", headers=token_header("fake-token"))
     assert response.status_code == 200, response.text
     response = client.get("/")
     assert response.status_code == 404, response.text
@@ -51,3 +61,173 @@ def test_base_url() -> None:
     assert response.status_code == 200, response.text
     response = client.get("/api/status")
     assert response.status_code == 404, response.text
+
+
+def test_skew_protection(edit_app: Starlette) -> None:
+    client = TestClient(edit_app)
+    # Unauthorized access
+    response = client.get("/")
+    assert response.status_code == 401, response.text
+    assert response.headers.get("Set-Cookie") is None
+
+    # Authorized access
+    response = client.get("/", headers=token_header("fake-token"))
+    assert response.status_code == 200, response.text
+    assert response.headers.get("Set-Cookie") is not None
+
+    # POST with a new passing skew protection token
+    response = client.post(
+        "/api/home/running_notebooks",
+        headers=token_header("fake-token"),
+    )
+    assert response.status_code == 200, response.text
+
+    # POST with a new skew protection token
+    response = client.post(
+        "/api/home/running_notebooks",
+        headers=token_header("fake-token", "old-skew-id"),
+    )
+    assert response.status_code == 401, response.text
+
+
+@pytest.fixture
+def edit_app() -> Starlette:
+    app = create_starlette_app(base_url="")
+    app.state.session_manager = get_mock_session_manager()
+    app.state.session_manager.mode = SessionMode.EDIT
+    app.state.config_manager = UserConfigManager()
+    # Mock out the server
+    uvicorn_server = uvicorn.Server(uvicorn.Config(app))
+    uvicorn_server.servers = []
+    app.state.server = uvicorn_server
+    app.state.host = "localhost"
+    app.state.port = 1234
+    app.state.base_url = ""
+    return app
+
+
+@pytest.fixture
+def read_app() -> Starlette:
+    app = create_starlette_app(base_url="")
+    app.state.session_manager = get_mock_session_manager()
+    app.state.session_manager.mode = SessionMode.RUN
+    app.state.config_manager = UserConfigManager()
+    # Mock out the server
+    uvicorn_server = uvicorn.Server(uvicorn.Config(app))
+    uvicorn_server.servers = []
+    app.state.server = uvicorn_server
+    app.state.host = "localhost"
+    app.state.port = 1234
+    app.state.base_url = ""
+    return app
+
+
+@pytest.fixture(params=["read_app", "edit_app"])
+def app(request: Any) -> Starlette:
+    return request.getfixturevalue(request.param)
+
+
+@pytest.fixture
+def no_auth_edit_app() -> Starlette:
+    app = create_starlette_app(base_url="", enable_auth=False)
+    app.state.session_manager = get_mock_session_manager()
+    app.state.session_manager.mode = SessionMode.EDIT
+    app.state.session_manager.auth_token = AuthToken("")  # no auth
+    app.state.config_manager = UserConfigManager()
+    # Mock out the server
+    uvicorn_server = uvicorn.Server(uvicorn.Config(app))
+    uvicorn_server.servers = []
+    app.state.server = uvicorn_server
+    app.state.host = "localhost"
+    app.state.port = 1234
+    app.state.base_url = ""
+    return app
+
+
+@pytest.fixture
+def no_auth_read_app() -> Starlette:
+    app = create_starlette_app(base_url="", enable_auth=False)
+    app.state.session_manager = get_mock_session_manager()
+    app.state.session_manager.mode = SessionMode.RUN
+    app.state.session_manager.auth_token = AuthToken("")  # no auth
+    app.state.config_manager = UserConfigManager()
+    # Mock out the server
+    uvicorn_server = uvicorn.Server(uvicorn.Config(app))
+    uvicorn_server.servers = []
+    app.state.server = uvicorn_server
+    app.state.host = "localhost"
+    app.state.port = 1234
+    app.state.base_url = ""
+    return app
+
+
+@pytest.fixture(params=["no_auth_read_app", "no_auth_edit_app"])
+def no_auth_app(request: Any) -> Starlette:
+    return request.getfixturevalue(request.param)
+
+
+class TestAuth:
+    def test_no_auth(self, app: Starlette) -> None:
+        # Test unauthorized access
+        client = TestClient(app)
+        response = client.get("/")
+        assert response.status_code == 401, response.text
+        assert response.headers.get("Set-Cookie") is None
+
+    def test_auth_by_query(self, app: Starlette) -> None:
+        # Test authorized access by auth_token in query
+        client = TestClient(app)
+        response = client.get("/?access_token=fake-token")
+        assert response.status_code == 200, response.text
+        assert response.headers.get("Set-Cookie") is not None
+
+        # Can access again with cookie
+        response = client.get("/")
+        assert response.status_code == 200, response.text
+
+    def tets_bath_auth_query(self, app: Starlette) -> None:
+        # Test unauthorized access with bad auth token
+        client = TestClient(app)
+        response = client.get("/?access_token=bad-token")
+        assert response.status_code == 401, response.text
+        assert response.headers.get("Set-Cookie") is None
+
+    def test_auth_by_header(self, app: Starlette) -> None:
+        # Test authorized access by auth_token in header (basic auth login)
+        client = TestClient(app)
+        response = client.get("/", headers=token_header("fake-token"))
+        assert response.status_code == 200, response.text
+        assert response.headers.get("Set-Cookie") is not None
+
+        # Can access again with cookie
+        response = client.get("/")
+        assert response.status_code == 200, response.text
+
+    def test_bad_auth_header(self, app: Starlette) -> None:
+        # Test unauthorized access with bad auth token
+        client = TestClient(app)
+        response = client.get("/", headers=token_header("bad-token"))
+        assert response.status_code == 401, response.text
+        assert response.headers.get("Set-Cookie") is None
+
+
+class TestNoAuth:
+    def test_no_auth(self, no_auth_app: Starlette) -> None:
+        client = TestClient(no_auth_app)
+        response = client.get("/")
+        assert response.status_code == 200, response.text
+        assert response.headers.get("Set-Cookie") is None
+
+    def tets_any_query(self, no_auth_app: Starlette) -> None:
+        # Test unauthorized access with bad auth token
+        client = TestClient(no_auth_app)
+        response = client.get("/?access_token=bad-token")
+        assert response.status_code == 200, response.text
+        assert response.headers.get("Set-Cookie") is None
+
+    def test_any_header(self, no_auth_app: Starlette) -> None:
+        # Test unauthorized access with bad auth token
+        client = TestClient(no_auth_app)
+        response = client.get("/", headers=token_header("bad-token"))
+        assert response.status_code == 200, response.text
+        assert response.headers.get("Set-Cookie") is None
diff --git a/tests/_server/conftest.py b/tests/_server/conftest.py
index 2badfe5d254..ffe1e8fe631 100644
--- a/tests/_server/conftest.py
+++ b/tests/_server/conftest.py
@@ -1,4 +1,6 @@
 # Copyright 2024 Marimo. All rights reserved.
+from __future__ import annotations
+
 import sys
 from typing import Generator, Iterator
 
@@ -12,7 +14,7 @@
 from marimo._server.utils import initialize_asyncio
 from tests._server.mocks import get_mock_session_manager
 
-app = create_starlette_app(base_url="")
+app = create_starlette_app(base_url="", enable_auth=True)
 
 
 @pytest.fixture(scope="session", autouse=True)
diff --git a/tests/_server/mocks.py b/tests/_server/mocks.py
index 362613b0df6..7d8e46b9993 100644
--- a/tests/_server/mocks.py
+++ b/tests/_server/mocks.py
@@ -1,6 +1,7 @@
 # Copyright 2024 Marimo. All rights reserved.
 from __future__ import annotations
 
+import base64
 import tempfile
 from typing import TYPE_CHECKING, Any, Callable, cast
 
@@ -8,6 +9,7 @@
 from marimo._server.file_router import AppFileRouter
 from marimo._server.model import SessionMode
 from marimo._server.sessions import NoopLspServer, SessionManager
+from marimo._server.tokens import AuthToken, SkewProtectionToken
 from marimo._utils.marimo_path import MarimoPath
 
 if TYPE_CHECKING:
@@ -53,8 +55,9 @@ def __():
         lsp_server=lsp_server,
         user_config_manager=UserConfigManager(),
         cli_args={},
+        auth_token=AuthToken("fake-token"),
     )
-    sm.server_token = "fake-token"
+    sm.skew_protection_token = SkewProtectionToken("skew-id-1")
     return sm
 
 
@@ -88,19 +91,21 @@ def with_session(
 
     def decorator(func: Callable[..., None]) -> Callable[..., None]:
         def wrapper(client: TestClient) -> None:
+            auth_token = get_session_manager(client).auth_token
+            headers = token_header(auth_token)
+
             with client.websocket_connect(
-                f"/ws?session_id={session_id}"
+                f"/ws?session_id={session_id}", headers=headers
             ) as websocket:
                 data = websocket.receive_text()
                 assert data
                 func(client)
             # shutdown after websocket exits, otherwise
             # test fails on Windows (loop closed twice)
-            server_token: str = get_session_manager(client).server_token
             if auto_shutdown:
                 client.post(
                     "/api/kernel/shutdown",
-                    headers={"Marimo-Server-Token": server_token},
+                    headers=headers,
                 )
 
         return wrapper
@@ -116,9 +121,10 @@ def with_read_session(
     def decorator(func: Callable[..., None]) -> Callable[..., None]:
         def wrapper(client: TestClient) -> None:
             session_manager = get_session_manager(client)
+            headers = token_header(session_manager.auth_token)
 
             with client.websocket_connect(
-                f"/ws?session_id={session_id}"
+                f"/ws?session_id={session_id}", headers=headers
             ) as websocket:
                 data = websocket.receive_text()
                 assert data
@@ -129,12 +135,21 @@ def wrapper(client: TestClient) -> None:
                 session_manager.mode = SessionMode.EDIT
             # shutdown after websocket exits, otherwise
             # test fails on Windows (loop closed twice)
-            server_token: str = session_manager.server_token
             client.post(
                 "/api/kernel/shutdown",
-                headers={"Marimo-Server-Token": server_token},
+                headers=headers,
             )
 
         return wrapper
 
     return decorator
+
+
+def token_header(
+    token: str | AuthToken = "fake-token", skew_id: str = "skew-id-1"
+) -> dict[str, str]:
+    encoded = base64.b64encode(f"marimo:{str(token)}".encode()).decode()
+    return {
+        "Authorization": f"Basic {encoded}",
+        "Marimo-Server-Token": skew_id,
+    }
diff --git a/tests/_server/templates/test_templates.py b/tests/_server/templates/test_templates.py
index ffb4c6751e3..c46811fd41c 100644
--- a/tests/_server/templates/test_templates.py
+++ b/tests/_server/templates/test_templates.py
@@ -10,6 +10,7 @@
 from marimo._messaging.cell_output import CellChannel, CellOutput
 from marimo._server.model import SessionMode
 from marimo._server.templates import templates
+from marimo._server.tokens import SkewProtectionToken
 from tests._server.templates.utils import normalize_index_html
 from tests.mocks import snapshotter
 
@@ -121,7 +122,7 @@ def setUp(self) -> None:
             self.html = f.read()
 
         self.user_config = DEFAULT_CONFIG
-        self.server_token = "token"
+        self.server_token = SkewProtectionToken("token")
         self.app_config = _AppConfig()
         self.filename = "notebook.py"
         self.code = "print('Hello, World!')"
diff --git a/tests/_server/test_session_manager.py b/tests/_server/test_session_manager.py
index ab7f6d01d1a..f7e90867c6d 100644
--- a/tests/_server/test_session_manager.py
+++ b/tests/_server/test_session_manager.py
@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import asyncio
 from unittest.mock import MagicMock, Mock
 
@@ -33,6 +35,7 @@ def session_manager():
         lsp_server=MagicMock(spec=LspServer),
         user_config_manager=UserConfigManager(),
         cli_args={},
+        auth_token=None,
     )
 
 


Code review comments:
diff_hunk: @@ -0,0 +1,67 @@
+# Authentication
+
+marimo provides a simple way to add token/password protection to your marimo server. Given that authentication is a complex topic, marimo does not provide a built-in authentication/authorization system, but instead makes it easy to add your own through ASGI middleware.
+
+## Enabling Basic Authentication
+
+Authentication is disabled by default, however this may change in the future. To enable authentication, you must pass `--token` to your `marimo edit/run/new` command from the Terminal. This will use a randomly generated token in `Edit mode` and a deterministic token in `Run mode` (based on the code of the notebook).
path: docs/guides/authentication.md
text: @user1:
@author Update this sentence to indicate that auth is enabled by default in `edit/tutorial/new` ?

diff_hunk: @@ -0,0 +1,190 @@
+# Copyright 2024 Marimo. All rights reserved.
+from __future__ import annotations
+
+import base64
+import secrets
+import typing
+from typing import TYPE_CHECKING, Any, Dict, Optional
+
+import starlette.status as status
+from starlette.datastructures import Secret
+from starlette.exceptions import HTTPException
+from starlette.middleware.sessions import SessionMiddleware
+from starlette.responses import JSONResponse
+
+from marimo import _loggers
+
+if TYPE_CHECKING:
+    from starlette.authentication import AuthenticationError
+    from starlette.requests import HTTPConnection
+    from starlette.types import ASGIApp, Receive, Scope, Send
+
+from marimo._server.api.deps import AppState
+
+LOGGER = _loggers.marimo_logger()
+TOKEN_QUERY_PARAM = "access_token"
+
+
+# Validates auth
+# - Checking for existing session cookie (already authenticated)
+# - Or authenticates by access_token in query params
+# - Or authenticates by basic auth
+def validate_auth(conn: HTTPConnection) -> bool:
+    state = AppState.from_app(conn.app)
+    auth_token = str(state.session_manager.auth_token)
+
+    # Check for session cookie
+    cookie_session = CookieSession(conn.session)
+    # Validate the cookie
+    if cookie_session.get_access_token() == auth_token:
+        return True  # Success
+
+    # Check for access_token
+    if TOKEN_QUERY_PARAM in conn.query_params:
+        # Validate the access_token
+        if conn.query_params[TOKEN_QUERY_PARAM] == auth_token:
+            LOGGER.debug("Validated access_token")
+            # Set the cookie
+            cookie_session.set_access_token(auth_token)
+            return True  # Success
+
+    # Check for basic auth
+    auth = conn.headers.get("Authorization")
+    if auth is not None:
+        username, password = _parse_basic_auth_header(auth)
+        if username and password == auth_token:
+            LOGGER.debug("Validated basic auth")
+            # Set the cookie
+            cookie_session.set_access_token(auth_token)
+            cookie_session.set_username(username)
+            return True  # Success
+
+    LOGGER.debug("Invalid auth")
+    return False
+
+
+def _parse_basic_auth_header(
+    header: str,
+) -> tuple[Optional[str], Optional[str]]:
+    scheme, _, credentials = header.partition(" ")
+
+    if scheme.lower() != "basic":
+        LOGGER.debug("Invalid auth scheme: %s", scheme)
+        return None, None
+
+    decoded = base64.b64decode(credentials).decode("utf-8")
+    username, _, password = decoded.partition(":")
+
+    if not password:
+        return None, None
+
+    LOGGER.debug("Validated basic auth for user: %s", username)
+
+    return username, password
+
+
+def raise_basic_auth_error() -> HTTPException:
+    return HTTPException(
+        status_code=status.HTTP_401_UNAUTHORIZED,
+        detail="Authorization header required",
+        headers={"WWW-Authenticate": "Basic"},
+    )
+
+
+def on_auth_error(
+    request: HTTPConnection, error: AuthenticationError
+) -> JSONResponse:
+    del request
+    del error
+    return JSONResponse(
+        status_code=status.HTTP_401_UNAUTHORIZED,
+        content={"detail": "Authorization header required"},
+        headers={"WWW-Authenticate": "Basic"},
+    )
+
+
+# This is random/new for each server instance
+RANDOM_SECRET = Secret(secrets.token_hex(32))
+
+
+class CookieSession:
+    """
+    Wrapper around starlette's Session to add typesafety
+    """
+
+    def __init__(self, session_state: Dict[str, Any]) -> None:
+        self.session_state = session_state
+
+    def get_access_token(self) -> str:
+        access_token: str = self.session_state.get("access_token", "")
+        return access_token
+
+    def get_username(self) -> str:
+        username: str = self.session_state.get("username", "")
+        return username
+
+    def set_access_token(self, token: str) -> None:
+        self.session_state["access_token"] = token
+
+    def set_username(self, username: str) -> None:
+        self.session_state["username"] = username
+
+
+class CustomSessionMiddleware(SessionMiddleware):
+    """
+    Wrapper around starlette's SessionMiddleware to:
+     - customize the session cookie based on the the port
+     - only run in Edit mode
+    """
+
+    def __init__(
+        self,
+        app: ASGIApp,
+        secret_key: typing.Union[str, Secret],
+        session_cookie: str = "session",
+        max_age: typing.Optional[int] = 14
+        * 24
+        * 60
+        * 60,  # 14 days, in seconds
+        path: str = "/",
+        same_site: typing.Literal["lax", "strict", "none"] = "lax",
+        https_only: bool = False,
+        domain: typing.Optional[str] = None,
+    ) -> None:
+        # We can't update the cookie here since
+        # we don't have access to the app state
+
+        self.original_session_cookie = session_cookie
+
+        super().__init__(
+            app,
+            secret_key,
+            session_cookie,
+            max_age,
+            path,
+            same_site,
+            https_only,
+            domain,
+        )
+
+    async def __call__(
+        self, scope: Scope, receive: Receive, send: Send
+    ) -> None:
+        if scope["type"] not in ("http", "websocket"):
+            await self.app(scope, receive, send)
+            return
+
+        state = AppState.from_app(scope["app"])
+        # disabled = state.session_manager.mode != SessionMode.EDIT
+        # if disabled:
+        #     return await self.app(scope, receive, send)
path: marimo/_server/api/auth.py
text: @user1:
Delete dead code?