ploneasgi.context

An authenticated Plone context, acquired from a worker thread.

A native ASGI route runs on the event loop, where there is no ZODB connection, no site, and no security manager. Anything that wants to touch content therefore has to establish all three — and doing that badly is how you write a file-disclosure bug. This module exists so it is written once, tested, and reused, instead of being re-derived per endpoint.

event loop      await run_authenticated(scope, work)
worker thread     open connection -> build request -> authenticate ->
                  setSite -> run `work` -> abort -> tear down
event loop      the return value

Everything Zope touches happens inside the worker thread and is torn down before it returns. A ZODB connection is not thread-safe and the security manager is thread-local, so neither may outlive the call — which is why the API takes a callable rather than handing back an open context.

What `work` may return. Plain data. Returning a persistent object hands the event loop an object whose connection has been closed; read a value out of it instead. There is no way to enforce this, so it is a rule, not a guarantee.

Credentials. Authentication is delegated to acl_users.validate(), never to parsing a cookie or JWT here. That keeps every PAS plugin working — including whatever a given site has installed — at the price of having to hand PAS a request that looks enough like a traversed one. See _publisher_request().

Transactions. Every call ends in transaction.abort(): a context reads and commits nothing. Writing is deliberately not enabled here; it needs its own answer to conflict errors and retries, which a read never has to give.

exception ploneasgi.context.NotAuthenticated

Raised when a route requires a real user and got Anonymous.

exception ploneasgi.context.SiteNotFound

Raised when the requested site path enters no Plone site.

class ploneasgi.context.PloneContext

What a unit of work is handed inside the worker thread.

Variables:

• portal -- The Plone site root, acquisition-wrapped and with a request.

• user -- The authenticated user, or Anonymous.

• request -- The synthesized publisher request.

property is_anonymous: bool

Returns:

True when no credentials authenticated.

require_user()

Assert that a real user authenticated.

Returns:

The authenticated user.

Raises:

NotAuthenticated -- When the request is anonymous.

Return type:

Any

traverse(path)

Traverse to an object, enforcing the current user's permissions.

Uses restrictedTraverse, which applies the security machinery at each step. Do not replace it with unrestrictedTraverse plus a permission check on the final object: that skips the checks on everything in between, which is the shape of most path-traversal disclosure bugs.

Parameters:

path (str) -- Path relative to the portal, e.g. "folder/document".

Returns:

The traversed object.

Raises:

• Unauthorized -- When the user may not reach it.

• KeyError -- When it does not exist.

Return type:

Any

check_permission(permission, obj)

Check a permission on an object as the current user.

Parameters:

• permission (str) -- Permission title, e.g. "View".

• obj (Any) -- The object to check against.

Returns:

Whether the permission is granted.

Return type:

bool

__init__(portal, user, request)

Parameters:

• portal (Any)

• user (Any)

• request (Any)

Return type:

None

async ploneasgi.context.run_authenticated(scope, work, *, site_id, db=None, limiter=None, body=None)

Run work in a worker thread with an authenticated Plone context.

Parameters:

• scope (MutableMapping[str, Any]) -- The ASGI connection scope; supplies the credentials.

• work (Callable[[PloneContext], Any]) -- Called with a PloneContext. Must return plain data — a persistent object outlives its connection and is unusable.

• site_id (str) -- Id of the Plone site to enter, or a path from the Zope root to it or to an object inside it (sites/Plone, Plone/folder) — the shape a VirtualHostMonster rewrite supplies.

• db (Any) -- ZODB database to use; defaults to Zope2.DB. Injectable so tests can pass the layer's DemoStorage, which is also how the worker-thread-sees-the-test-database question gets answered.

• limiter (CapacityLimiter | None) -- Bounds concurrent calls. Defaults to the one ploneasgi.app.create_app() publishes as app.state.zodb_limiter — the adapter's app_limiter — so native routes share the WSGI requests' ZODB pool budget; with neither, anyio's default pool.

• body (IO[bytes] | None) -- The request body, already read on the event loop and positioned at its start, for ctx.request to read. The caller owns it and closes it; reading it here, in the thread, would mean a blocking round trip to the event loop per read.

Returns:

Whatever work returns.

Return type:

Any

ploneasgi.context.limiter_for(scope)

The ZODB limiter of the application serving a request.

Starlette puts the application on every scope it serves, which is how a route built far from ploneasgi.app.create_app() — by the ZCML directive, say — reaches the limiter without being handed it.

Parameters:

scope (MutableMapping[str, Any]) -- The ASGI connection scope.

Returns:

app.state.zodb_limiter, or None when the scope carries no application or the application has none.

Return type:

CapacityLimiter | None

ploneasgi.context.default_db()

The process-wide ZODB set up by Zope's startup.

A function rather than a direct Zope2.DB read at the call site so that a caller which cannot pass db= explicitly — a route built by the ZCML directive, for instance — still has a single place to redirect in tests.

Returns:

The ZODB database.

Raises:

RuntimeError -- When Zope has not been started in this process.

Return type:

Any