Built-in middleware#
CORS#
CORS (Cross-Origin Resource Sharing) is a common security
mechanism that is often implemented using middleware. To enable CORS in a starlite application simply pass an instance
of CORSConfig
to the Starlite constructor
:
from starlite import CORSConfig, Starlite
cors_config = CORSConfig(allow_origins=["https://www.example.com"])
app = Starlite(route_handlers=[...], cors_config=cors_config)
CSRF#
CSRF (Cross-site request forgery) is a type of attack where unauthorized commands are submitted from a user that the web application trusts. This attack often uses social engineering that tricks the victim into clicking a URL that contains a maliciously crafted, unauthorized request for a particular Web application. The user’s browser then sends this maliciously crafted request to the targeted Web application. If the user is in an active session with the Web application, the application treats this new request as an authorized request submitted by the user. Thus, the attacker can force the user to perform an action the user didn’t intend, for example:
POST /send-money HTTP/1.1
Host: target.web.app
Content-Type: application/x-www-form-urlencoded
amount=1000usd&to=attacker@evil.com
This middleware prevents CSRF attacks by doing the following:
On the first “safe” request (e.g GET) - set a cookie with a special token created by the server
- On each subsequent “unsafe” request (e.g POST) - make sure the request contains either a
form field or an additional header that has this token
To enable CSRF protection in a Starlite application simply pass an instance of
CSRFConfig
to the Starlite constructor:
from starlite import Starlite, CSRFConfig
csrf_config = CSRFConfig(secret="my-secret")
app = Starlite(route_handlers=[...], csrf_config=csrf_config)
Routes can be marked as being exempt from the protection offered by this middleware via handler opts
from starlite import post
@post("/post", exclude_from_csrf=True)
def handler() -> None:
...
If you need to exempt many routes at once you might want to consider using the
exclude
kwarg which accepts list of path
patterns to skip in the middleware.
Allowed Hosts#
Another common security mechanism is to require that each incoming request has a “Host” or “X-Forwarded-Host” header, and then to restrict hosts to a specific set of domains - what’s called “allowed hosts”.
Starlite includes an AllowedHostsMiddleware
class that can be
easily enabled by either passing an instance of AllowedHostsConfig
or a
list of domains to the Starlite constructor
:
from starlite import Starlite, AllowedHostsConfig
app = Starlite(
route_handlers=[...],
allowed_hosts=AllowedHostsConfig(
allowed_hosts=["*.example.com", "www.wikipedia.org"]
),
)
Note
You can use wildcard prefixes (*.
) in the beginning of a domain to match any combination of subdomains. Thus,
*.example.com
will match www.example.com
but also x.y.z.example.com
etc. You can also simply put *
in trusted hosts, which means allow all. This is akin to turning the middleware off, so in this case it may be
better to not enable it in the first place. You should note that a wildcard can only be used only in the prefix of a
domain name, not in the middle or end. Doing so will result in a validation exception being raised.
Compression#
HTML responses can optionally be compressed. Starlite has built in support for gzip and brotli. Gzip support is provided
through the built-in Starlette classes, and brotli support can be added by installing the brotli
extras.
You can enable either backend by passing an instance of CompressionConfig
into the compression_config
the Starlite constructor
.
GZIP#
You can enable gzip compression of responses by passing an instance of starlite.config.CompressionConfig
with
the backend
parameter set to "gzip"
.
You can configure the following additional gzip-specific values:
minimum_size
: the minimum threshold for response size to enable compression. Smaller responses will not becompressed. Defaults is
500
, i.e. half a kilobyte.
gzip_compress_level
: a range between 0-9, see the official python docs.Defaults to
9
, which is the maximum value.
from starlite import Starlite, CompressionConfig
app = Starlite(
route_handlers=[...],
compression_config=CompressionConfig(backend="gzip", gzip_compress_level=9),
)
Brotli#
The Brotli package is required to run this middleware. It is available as an extras to starlite with the brotli
extra (pip install starlite[brotli]
).
You can enable brotli compression of responses by passing an instance of starlite.config.CompressionConfig
with
the backend
parameter set to "brotli"
.
You can configure the following additional brotli-specific values:
minimum_size
: the minimum threshold for response size to enable compression. Smaller responses will not becompressed. Defaults is
500
, i.e. half a kilobyte.
brotli_quality
: Range [0-11], Controls the compression-speed vs compression-density tradeoff. The higher thequality, the slower the compression.
brotli_mode
: The compression mode can be MODE_GENERIC (default), MODE_TEXT (for UTF-8 format text input) orMODE_FONT (for WOFF 2.0).
brotli_lgwin
: Base 2 logarithm of size. Range is 10 to 24. Defaults to 22.brotli_lgblock
: Base 2 logarithm of the maximum input block size. Range is 16 to 24. If set to 0, the value willbe set based on the quality. Defaults to 0.
brotli_gzip_fallback
: a boolean to indicate if gzip should be used if brotli is not supported.
from starlite import Starlite
from starlite.config import CompressionConfig
app = Starlite(
route_handlers=[...],
compression_config=CompressionConfig(backend="brotli", brotli_gzip_fallback=True),
)
Rate-Limit Middleware#
Starlite includes an optional RateLimitMiddleware
that follows
the IETF RateLimit draft specification.
To use the rate limit middleware, use the RateLimitConfig
:
from starlite import MediaType, Starlite, get
from starlite.middleware import RateLimitConfig
rate_limit_config = RateLimitConfig(rate_limit=("minute", 1), exclude=["/schema"])
@get("/", media_type=MediaType.TEXT)
def handler() -> str:
"""Handler which should not be accessed more than once per minute."""
return "ok"
app = Starlite(route_handlers=[handler], middleware=[rate_limit_config.middleware])
The only required configuration kwarg is rate_limit
, which expects a tuple containing a time-unit (second
,
minute
, hour
, day
) and a value for the request quota (integer). For the other configuration options.
Logging Middleware#
Starlite ships with a robust logging middleware that allows logging HTTP request and responses while building on the app level logging configuration:
from typing import Dict
from starlite import LoggingConfig, Starlite, get
from starlite.middleware import LoggingMiddlewareConfig
logging_middleware_config = LoggingMiddlewareConfig()
@get("/")
def my_handler() -> Dict[str, str]:
return {"hello": "world"}
app = Starlite(
route_handlers=[my_handler],
logging_config=LoggingConfig(),
middleware=[logging_middleware_config.middleware],
)
from starlite import LoggingConfig, Starlite, get
from starlite.middleware import LoggingMiddlewareConfig
logging_middleware_config = LoggingMiddlewareConfig()
@get("/")
def my_handler() -> dict[str, str]:
return {"hello": "world"}
app = Starlite(
route_handlers=[my_handler],
logging_config=LoggingConfig(),
middleware=[logging_middleware_config.middleware],
)
The logging middleware uses the logger configuration defined on the application level, which allows for using both stdlib logging or structlog , depending on the configuration used (see app level logging configuration for more details).
Obfuscating Logging Output#
Sometimes certain data, e.g. request or response headers, needs to be obfuscated. This is supported by the middleware configuration:
from starlite.middleware import LoggingMiddlewareConfig
logging_middleware_config = LoggingMiddlewareConfig(
request_cookies_to_obfuscate={"my-custom-session-key"},
response_cookies_to_obfuscate={"my-custom-session-key"},
request_headers_to_obfuscate={"my-custom-header"},
response_headers_to_obfuscate={"my-custom-header"},
)
The middleware will obfuscate the headers Authorization
and X-API-KEY
, and the cookie session
by default.
Compression and Logging of Response Body#
If both CompressionConfig
and
LoggingMiddleware
have been defined for the application, the response
body will be omitted from response logging if it has been compressed, even if "body"
has been included in
response_log_fields
. To force the body of
compressed responses to be logged, set
include_compressed_body
to True
, in
addition to including "body"
in response_log_fields
.
Session Middleware#
Starlite includes a SessionMiddleware
,
offering client- and server-side sessions. Different storage mechanisms are available through
SessionBackends
, and include support for
storing data in:
Setting up the middleware#
To start using sessions in your application all you have to do is create an instance
of a configuration
object and
add its middleware to your application’s middleware stack:
from os import urandom
from typing import Dict
from starlite import Request, Starlite, delete, get, post
from starlite.middleware.session.cookie_backend import CookieBackendConfig
# we initialize to config with a 16 byte key, i.e. 128 a bit key.
# in real world usage we should inject the secret from the environment
session_config = CookieBackendConfig(secret=urandom(16)) # type: ignore[arg-type]
@get("/session")
def check_session_handler(request: Request) -> Dict[str, bool]:
"""Handler function that accesses request.session."""
return {"has_session": request.session != {}}
@post("/session")
def create_session_handler(request: Request) -> None:
"""Handler to set the session."""
if not request.session:
# value can be a dictionary or pydantic model
request.set_session({"username": "moishezuchmir"})
@delete("/session")
def delete_session_handler(request: Request) -> None:
"""Handler to clear the session."""
if request.session:
request.clear_session()
app = Starlite(
route_handlers=[check_session_handler, create_session_handler, delete_session_handler],
middleware=[session_config.middleware],
)
from os import urandom
from starlite import Request, Starlite, delete, get, post
from starlite.middleware.session.cookie_backend import CookieBackendConfig
# we initialize to config with a 16 byte key, i.e. 128 a bit key.
# in real world usage we should inject the secret from the environment
session_config = CookieBackendConfig(secret=urandom(16)) # type: ignore[arg-type]
@get("/session")
def check_session_handler(request: Request) -> dict[str, bool]:
"""Handler function that accesses request.session."""
return {"has_session": request.session != {}}
@post("/session")
def create_session_handler(request: Request) -> None:
"""Handler to set the session."""
if not request.session:
# value can be a dictionary or pydantic model
request.set_session({"username": "moishezuchmir"})
@delete("/session")
def delete_session_handler(request: Request) -> None:
"""Handler to clear the session."""
if request.session:
request.clear_session()
app = Starlite(
route_handlers=[check_session_handler, create_session_handler, delete_session_handler],
middleware=[session_config.middleware],
)
Note
Since both client- and server-side sessions rely on cookies (one for storing the actual session
data, the other for storing the session ID), they share most of the cookie configuration.
A complete reference of the cookie configuration can be found at BaseBackendConfig
.
Client-side sessions#
Client side sessions are available through the CookieBackend
,
which offers strong AES-CGM encryption security best practices while support cookie splitting.
Important
CookieBackend
requires the cryptography library,
which can be installed together with starlite as an extra using pip install starlite[cryptography]
from os import urandom
from starlite import Starlite
from starlite.middleware.session.cookie_backend import CookieBackendConfig
session_config = CookieBackendConfig(secret=urandom(16)) # type: ignore[arg-type]
app = Starlite(route_handlers=[], middleware=[session_config.middleware])
See also
Server-side sessions#
Server side session store data - as the name suggests - on the server instead of the client. They use a cookie containing a session ID which is a randomly generated string to identify a client and load the appropriate data from the storage backend.
File storage#
The FileBackend
will store session data
in files on disk, alongside some metadata. Files containing expired sessions will only be deleted
when trying to access them. Expired session files can be manually deleted using the
delete_expired
method.
from pathlib import Path
from starlite import Starlite
from starlite.middleware.session.file_backend import FileBackendConfig
session_config = FileBackendConfig(storage_path=Path("/path/to/session/storage"))
app = Starlite(route_handlers=[], middleware=[session_config.middleware])
Redis storage#
The Redis backend
can store session data
in redis. Session data stored in redis will expire automatically after its
max_age
has been passed.
Important
This requires the redis
package. To install it you can install starlite with
pip install starlite[redis]
from redis.asyncio import Redis
from starlite import Starlite
from starlite.middleware.session.redis_backend import RedisBackendConfig
session_config = RedisBackendConfig(redis=Redis(host="localhost", port=6379, db=0))
app = Starlite(route_handlers=[], middleware=[session_config.middleware])
Memcached storage#
The Memcached backend
can store session data
in memcached. Session data stored in memcached will expire automatically after its
max_age
has been passed.
Important
This requires the aiomemcache
package. To install it you can install starlite with
pip install starlite[memcached]
from aiomcache import Client
from starlite import Starlite
from starlite.middleware.session.memcached_backend import MemcachedBackendConfig
session_config = MemcachedBackendConfig(memcached=Client("127.0.0.1"))
app = Starlite(route_handlers=[], middleware=[session_config.middleware])
In-memory storage#
The Memory backend
can store
session data in memory.
Important
This should not be used in production. It primarily exists as a dummy backend for testing purposes. It is not process safe, and data will not be persisted.
from starlite import Starlite
from starlite.middleware.session.memory_backend import MemoryBackendConfig
session_config = MemoryBackendConfig()
app = Starlite(route_handlers=[], middleware=[session_config.middleware])
Database storage#
Database storage is currently offered through the
SQLAlchemyBackend
.
It supports both sync and async-engines and integrates with the
SQLAlchemyPlugin. Expired sessions will only be deleted when trying to access them.
They can be manually deleted using the
delete_expired
method.
There are two backends for SQLAlchemy:
SQLAlchemyBackend
for synchronous enginesAsyncSQLAlchemyBackend
for asynchronous engines
When using the configuration
object,
it will automatically pick the correct backend to use based on the engine configuration.
Important
This requires sqlalchemy. You can install it via
pip install sqlalchemy
.
from sqlalchemy.orm import declarative_base
from starlite import Starlite
from starlite.middleware.session.sqlalchemy_backend import (
SQLAlchemyBackendConfig,
create_session_model,
)
from starlite.plugins.sql_alchemy import SQLAlchemyConfig, SQLAlchemyPlugin
Base = declarative_base()
sqlalchemy_config = SQLAlchemyConfig(connection_string="sqlite+pysqlite://", use_async_engine=False)
sqlalchemy_plugin = SQLAlchemyPlugin(config=sqlalchemy_config)
SessionModel = create_session_model(Base)
session_config = SQLAlchemyBackendConfig(plugin=sqlalchemy_plugin, model=SessionModel)
def on_startup() -> None:
"""Initialize the database."""
Base.metadata.create_all(sqlalchemy_config.engine) # type: ignore
app = Starlite(
route_handlers=[],
middleware=[session_config.middleware],
plugins=[sqlalchemy_plugin],
on_startup=[on_startup],
)
from sqlalchemy.orm import declarative_base
from starlite import Starlite
from starlite.middleware.session.sqlalchemy_backend import (
SQLAlchemyBackendConfig,
create_session_model,
)
from starlite.plugins.sql_alchemy import SQLAlchemyConfig, SQLAlchemyPlugin
Base = declarative_base()
SessionModel = create_session_model(Base)
sqlalchemy_config = SQLAlchemyConfig(connection_string="sqlite+aiosqlite://")
sqlalchemy_plugin = SQLAlchemyPlugin(config=sqlalchemy_config)
session_config = SQLAlchemyBackendConfig(
plugin=sqlalchemy_plugin,
model=SessionModel,
)
async def on_startup() -> None:
"""Initialize the database."""
async with sqlalchemy_config.engine.begin() as conn: # type: ignore
await conn.run_sync(Base.metadata.create_all) # pyright: ignore
app = Starlite(
route_handlers=[], middleware=[session_config.middleware], plugins=[sqlalchemy_plugin], on_startup=[on_startup]
)
Supplying your own session model#
If you wish to extend the built-in session model, you can mixin the
SessionModelMixin
into your own classes:
from sqlalchemy import Column, Integer, String
from sqlalchemy.orm import declarative_base
from starlite import Starlite
from starlite.middleware.session.sqlalchemy_backend import (
SessionModelMixin,
SQLAlchemyBackendConfig,
)
from starlite.plugins.sql_alchemy import SQLAlchemyConfig, SQLAlchemyPlugin
Base = declarative_base()
class SessionModel(Base, SessionModelMixin): # pyright: ignore [reportGeneralTypeIssues]
__tablename__ = "my-session-table"
id = Column(Integer, primary_key=True)
additional_data = Column(String)
sqlalchemy_config = SQLAlchemyConfig(connection_string="sqlite+aiosqlite://")
sqlalchemy_plugin = SQLAlchemyPlugin(config=sqlalchemy_config)
session_config = SQLAlchemyBackendConfig(
plugin=sqlalchemy_plugin,
model=SessionModel,
)
async def on_startup() -> None:
"""Initialize the database."""
async with sqlalchemy_config.engine.begin() as conn: # type: ignore
await conn.run_sync(Base.metadata.create_all) # pyright: ignore
app = Starlite(
route_handlers=[], middleware=[session_config.middleware], plugins=[sqlalchemy_plugin], on_startup=[on_startup]
)
See also
Accessing the storage backend directly#
In some situations you might want to access the storage backend directly, outside a
request. For example to delete a specific session’s data, or delete expired sessions
from the database when using the SQLAlchemyBackend
.
from pathlib import Path
from starlite import Starlite
from starlite.middleware.session.file_backend import FileBackend, FileBackendConfig
session_config = FileBackendConfig(storage_path=Path("/path/to/session/storage"))
session_backend = FileBackend(config=session_config)
async def clear_expired_sessions() -> None:
"""Delete all expired sessions."""
await session_backend.delete_expired()
app = Starlite(route_handlers=[], middleware=[session_backend.config.middleware])