> ## Documentation Index
> Fetch the complete documentation index at: https://docs.activeviam.com/llms.txt
> Use this file to discover all available pages before exploring further.

# atoti.proxy.Proxy

### *final class* atoti.proxy.Proxy

The proxy alllowing to forward requests from Atoti Server to another server.

A request made to `f"{session.url}/proxy/some/path?foo=bar"` will be proxied to `f"{session.proxy.url}/some/path?foo=bar"`:

The new request will have an Authorization HTTP header containing a JWT.
This JWT contains the following claims:

* `sub`: The name of the user making the request to the proxy.
* `authorities`: The list of roles of the user.

This JWT is signed by Atoti Server, except when using the [community edition](../guides/unlocking_all_features).

<Warning>
  For maximal flexibility, `f"{session.url}/proxy"` does not require authentication.
  It is the responsibility of the server based at [`url`](./atoti.proxy.Proxy.url#atoti.proxy.Proxy.url) to handle authentication and authorization if needed.
</Warning>

```mermaid theme={"languages":{"custom":["/engine/python-sdk/0.9/languages/pycon.tmLanguage.json"]}}
sequenceDiagram
  participant C as Client
  participant S as Atoti Server<br>https://example.com:1337
  participant P as Proxy target server<br>https://custom.com:1991
  C->>S: session.proxy.url = "https://custom.com:1991"
  C->>S: POST /proxy/some/path?foo=bar
  S->>P: POST /some/path?foo=bar<br>with extra HTTP headers:<br>• Authorization: Bearer $jwt<br>• X-Forwarded-Proto: https<br>• X-Forwarded-Host: example.com:1337
  P->>P: ⚠ Check $jwt
  P->>S: HTTP $status_code response
  S->>C: HTTP $status_code response
```

### Example

Using a custom JWT key pair to be able to verify JWT signatures in the local server below:

```pycon theme={"languages":{"custom":["/engine/python-sdk/0.9/languages/pycon.tmLanguage.json"]}}
>>> public_key, public_key_pem, private_key = (
...     (jwt_directory / filename).read_text()
...     for filename in ["public-key.txt", "public-key.pem", "private-key.txt"]
... )
```

```pycon theme={"languages":{"custom":["/engine/python-sdk/0.9/languages/pycon.tmLanguage.json"]}}
>>> from jwt import decode
>>> def get_user(jwt: str, /) -> tuple[str, list[str]]:
...     """Check the JWT signature and return the user name and roles claims.
...
...     Verifying the JWT signature is primordial to ensure
...     that the request is legitimate and coming from Atoti Server.
...
...     This function does not rely on Atoti Python SDK
...     and could actually be implemented in any language.
...     """
...     claims = decode(jwt, algorithms=["RS512"], key=public_key_pem)
...     user_name = claims["sub"]
...     user_roles = claims["authorities"]
...     return user_name, user_roles
```

Starting a secured session:

```pycon theme={"languages":{"custom":["/engine/python-sdk/0.9/languages/pycon.tmLanguage.json"]}}
>>> session_config = tt.SessionConfig(
...     security=tt.SecurityConfig(
...         jwt=tt.JwtConfig(key_pair=tt.KeyPair(public_key, private_key))
...     )
... )
>>> secured_session = tt.Session.start(session_config)
```

The proxy is disabled by default:

```pycon theme={"languages":{"custom":["/engine/python-sdk/0.9/languages/pycon.tmLanguage.json"]}}
>>> import httpx
>>> print(secured_session.proxy.url)
None
>>> response = httpx.get(f"{secured_session.url}/proxy")
>>> response.status_code
501
>>> response.text
'The proxy is disabled.'
```

Starting a local server:

```pycon theme={"languages":{"custom":["/engine/python-sdk/0.9/languages/pycon.tmLanguage.json"]}}
>>> from json import dumps
>>> from http import HTTPStatus
>>> from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
>>> from threading import Thread
>>> from urllib.parse import parse_qs, urlparse
```

```pycon theme={"languages":{"custom":["/engine/python-sdk/0.9/languages/pycon.tmLanguage.json"]}}
>>> class RequestHandler(BaseHTTPRequestHandler):
...     protocol_version = "HTTP/1.1"
...
...     def do_GET(self) -> None:
...         body = b""
...         http_status = HTTPStatus.OK
...         parse_result = urlparse(self.path)
...         match parse_result.path, parse_qs(parse_result.query):
...             case "/whoami", {"mode": [mode]}:
...                 token_type, jwt = self.headers["Authorization"].split(" ")
...                 match mode:
...                     case "jwt":
...                         # Most projects will want to raise an error if
...                         # `ROLE_USER` is not included in `user_roles`.
...                         user_name, user_roles = get_user(jwt)
...                     case "session":
...                         # This server does not have to be aware of the Atoti session URL
...                         # since it can be reconstructed from headers passed by the proxy.
...                         protocol = self.headers["X-Forwarded-Proto"]
...                         host = self.headers["X-Forwarded-Host"]
...                         session_url = f"{protocol}://{host}"
...                         authentication = tt.TokenAuthentication(
...                             jwt, token_type=token_type
...                         )
...                         try:
...                             with tt.Session.connect(
...                                 session_url, authentication=authentication
...                             ) as session:
...                                 user = session.user
...                             user_name = user.name
...                             user_roles = sorted(user.roles)
...                         except Exception:
...                             http_status = HTTPStatus.FORBIDDEN
...                     case _:
...                         http_status = HTTPStatus.BAD_REQUEST
...                 if http_status == HTTPStatus.OK:
...                     body = dumps(
...                         {"name": user_name, "roles": user_roles}
...                     ).encode("utf-8")
...             case _:
...                 http_status = HTTPStatus.NOT_FOUND
...
...         self.send_response_only(http_status)
...         self.send_header("Content-Length", str(len(body)))
...         self.end_headers()
...         if body:
...             self.wfile.write(body)
...
...     def do_POST(self) -> None:
...         content_length = int(self.headers["Content-Length"])
...         body = self.rfile.read(content_length)
...         self.send_response_only(HTTPStatus.OK)
...         self.send_header("Content-Length", str(content_length))
...         # To show that response headers are sent back to the client.
...         self.send_header("X-Custom-Header", "lorem ipsum")
...         self.end_headers()
...         self.wfile.write(body)
```

```pycon theme={"languages":{"custom":["/engine/python-sdk/0.9/languages/pycon.tmLanguage.json"]}}
>>> local_server = ThreadingHTTPServer(("localhost", 0), RequestHandler)
>>> thread = Thread(daemon=True, target=local_server.serve_forever)
>>> thread.start()
```

The server above is implemented in Python but the proxy target can be any HTTP server or even a lambda function in the cloud.

Configuring the proxy to forward requests to the local server:

```pycon theme={"languages":{"custom":["/engine/python-sdk/0.9/languages/pycon.tmLanguage.json"]}}
>>> secured_session.proxy.url = f"http://localhost:{local_server.server_port}"
```

Calling the proxy without authentication:

```pycon theme={"languages":{"custom":["/engine/python-sdk/0.9/languages/pycon.tmLanguage.json"]}}
>>> response = httpx.get(f"{secured_session.url}/proxy/whoami?mode=jwt")
>>> response.status_code
200
>>> response.json()
{'name': 'anonymousUser', 'roles': ['ROLE_ANONYMOUS']}
```

Without authentication, ROLE\_USER is missing and thus [`atoti.Session.connect()`](./atoti.Session.connect#atoti.Session.connect) will raise an error:

```pycon theme={"languages":{"custom":["/engine/python-sdk/0.9/languages/pycon.tmLanguage.json"]}}
>>> response = httpx.get(f"{secured_session.url}/proxy/whoami?mode=session")
>>> response.status_code
403
```

Adding a user to the session and calling the proxy as this user:

```pycon theme={"languages":{"custom":["/engine/python-sdk/0.9/languages/pycon.tmLanguage.json"]}}
>>> username, password = "Alice", "abcdef123456"
>>> alice_roles = {"ROLE_USER", "ROLE_WONDERLAND"}
>>> secured_session.security.individual_roles[username] = alice_roles
>>> secured_session.security.basic_authentication.credentials[username] = (
...     password
... )
>>> response = httpx.get(
...     f"{secured_session.url}/proxy/whoami?mode=jwt",
...     auth=(username, password),
... )
>>> response.status_code
200
>>> response.json()
{'name': 'Alice', 'roles': ['ROLE_USER', 'ROLE_WONDERLAND']}
```

ROLE\_USER is present so [`atoti.Session.connect()`](./atoti.Session.connect#atoti.Session.connect) and the access to [`atoti.Session.user`](./atoti.Session.user#atoti.Session.user) will both succeed:

```pycon theme={"languages":{"custom":["/engine/python-sdk/0.9/languages/pycon.tmLanguage.json"]}}
>>> response = httpx.get(
...     f"{secured_session.url}/proxy/whoami?mode=session",
...     auth=(username, password),
... )
>>> response.status_code
200
>>> response.json()
{'name': 'Alice', 'roles': ['ROLE_USER', 'ROLE_WONDERLAND']}
```

Trying an unsupported mode:

```pycon theme={"languages":{"custom":["/engine/python-sdk/0.9/languages/pycon.tmLanguage.json"]}}
>>> response = httpx.get(f"{secured_session.url}/proxy/whoami?mode=invalid")
>>> response.status_code
400
```

Trying an unhandled path:

```pycon theme={"languages":{"custom":["/engine/python-sdk/0.9/languages/pycon.tmLanguage.json"]}}
>>> response = httpx.get(f"{secured_session.url}/proxy")
>>> response.status_code
404
```

Request and response bodies are not limited to strings:

```pycon theme={"languages":{"custom":["/engine/python-sdk/0.9/languages/pycon.tmLanguage.json"]}}
>>> from random import randbytes
>>> bytes = randbytes(42 * 1024 * 1024)
>>> response = httpx.post(f"{secured_session.url}/proxy", content=bytes)
>>> response.status_code
200
>>> response.headers["Content-Length"]
'44040192'
>>> response.headers["X-Custom-Header"]
'lorem ipsum'
>>> response.content == bytes
True
```

Disabling the proxy:

```pycon theme={"languages":{"custom":["/engine/python-sdk/0.9/languages/pycon.tmLanguage.json"]}}
>>> secured_session.proxy.url = None
>>> httpx.get(f"{secured_session.url}/proxy").status_code
501
```

On a session without security, [`atoti.Session.connect()`](./atoti.Session.connect#atoti.Session.connect) always succeeds, even for anonymous calls:

```pycon theme={"languages":{"custom":["/engine/python-sdk/0.9/languages/pycon.tmLanguage.json"]}}
>>> print(session.proxy.url)
None
>>> session.proxy.url = f"http://localhost:{local_server.server_port}"
>>> response = httpx.get(f"{session.url}/proxy/whoami?mode=session")
>>> response.status_code
200
>>> response.json()
{'name': 'anonymousUser', 'roles': ['ROLE_ADMIN', 'ROLE_ANONYMOUS', 'ROLE_USER']}
```

See [`atoti.Session.user`](./atoti.Session.user#atoti.Session.user) for an explanation of the roles above.

Stopping the local server:

```pycon theme={"languages":{"custom":["/engine/python-sdk/0.9/languages/pycon.tmLanguage.json"]}}
>>> local_server.shutdown()
>>> local_server.server_close()
>>> thread.join()
```

<Callout icon="link">
  **See also**:

  * [`atoti.Session.endpoint()`](./atoti.Session.endpoint#atoti.Session.endpoint).
</Callout>

| [`url`](./atoti.proxy.Proxy.url#atoti.proxy.Proxy.url) | The URL towards which requests are forwarded. |
| ------------------------------------------------------ | --------------------------------------------- |