Skip to content

5. Context

What You'll Learn

  • Pass authentication context from transport layer
  • Use hierarchical context validation
  • Implement role-based access control

Basic Context

basic_context.py
from dataclasses import dataclass
import uuid
from jsonrpc import JSONRPC, Method
from jsonrpc.errors import JSONRPCError

class Unauthenticated(JSONRPCError):
    code = -32010
    message = 'Authentication required'


@dataclass
class RequestContext:
    request_id: str
    user_id: int | None = None

@dataclass
class GetProfileParams:
    include_email: bool = False

@dataclass
class ProfileResult:
    user_id: int
    username: str
    email: str | None

class GetProfile(Method):
    def execute(self, params: GetProfileParams, context: RequestContext) -> ProfileResult:
        if not context.user_id:
            raise Unauthenticated()

        return ProfileResult(
            user_id=context.user_id,
            username=f"user_{context.user_id}",
            email="user@example.com" if params.include_email else None,
        )

rpc = JSONRPC(version='2.0', context_type=RequestContext)
rpc.register('get_profile', GetProfile())

ctx = RequestContext(request_id=str(uuid.uuid4()), user_id=42)
request = '{"jsonrpc": "2.0", "method": "get_profile", "params": {"include_email": true}, "id": 1}'
response = rpc.handle(request, context=ctx)

Response:

response.json
{
  "jsonrpc": "2.0",
  "result": {
    "user_id": 42,
    "username": "user_42",
    "email": "user@example.com"
  },
  "id": 1
}

Role-Based Access Control

rbac.py
from dataclasses import dataclass
from jsonrpc import JSONRPC, MethodGroup, Method
from jsonrpc.errors import JSONRPCError

class Forbidden(JSONRPCError):
    """Authenticated, but not allowed. Distinct from Unauthenticated on purpose."""

    code = -32011
    message = 'Forbidden'


@dataclass
class AuthContext:
    user_id: int
    is_admin: bool = False
    is_moderator: bool = False

@dataclass
class DeleteUserParams:
    target_user_id: int

@dataclass
class BanUserParams:
    user_id: int
    reason: str
    duration_days: int

@dataclass
class DeleteUserResult:
    deleted_user_id: int
    deleted_by: int

@dataclass
class BanUserResult:
    banned_user_id: int
    reason: str
    banned_by: int

class DeleteUser(Method):
    def execute(self, params: DeleteUserParams, context: AuthContext) -> DeleteUserResult:
        if not context.is_admin:
            raise Forbidden("Admin privileges required")
        return DeleteUserResult(
            deleted_user_id=params.target_user_id,
            deleted_by=context.user_id,
        )

class BanUser(Method):
    def execute(self, params: BanUserParams, context: AuthContext) -> BanUserResult:
        if not (context.is_admin or context.is_moderator):
            raise Forbidden("Moderator or admin privileges required")
        return BanUserResult(
            banned_user_id=params.user_id,
            reason=params.reason,
            banned_by=context.user_id,
        )

rpc = JSONRPC(version='2.0', context_type=AuthContext)

admin = MethodGroup()
admin.register('delete_user', DeleteUser())
rpc.register('admin', admin)

moderation = MethodGroup()
moderation.register('ban_user', BanUser())
rpc.register('moderation', moderation)

Admin request (allowed):

admin_request.json
{
  "jsonrpc": "2.0",
  "method": "admin.delete_user",
  "params": {"target_user_id": 999},
  "id": 1
}

ctx = AuthContext(user_id=1, is_admin=True)
response = rpc.handle(request, context=ctx)

Non-admin request (denied):

ctx = AuthContext(user_id=42, is_admin=False)
response = rpc.handle(request, context=ctx)

Error response:

access_denied.json
{
  "jsonrpc": "2.0",
  "error": {
    "code": -32011,
    "message": "Admin privileges required"
  },
  "id": 1
}

Flask Integration

X-User-ID is not authentication

These examples read identity straight off a request header, because the subject here is context, not credentials — a header keeps the example to one line. A caller can set any header they like, so an endpoint wired this way grants whatever the caller asserts, and an audit trail built on it names whoever the caller named.

In real code, derive user_id from something you verified: a signed session, a validated JWT claim, a token looked up in your store. See Middleware → Transport layer for what that looks like. The RPC layer cannot check this for you: it never sees your headers, and a guard like RequireAuthGroup can only trust the context it is handed.

flask_context.py
from flask import Flask, request
from dataclasses import dataclass
from jsonrpc import JSONRPC, Method
from jsonrpc.errors import JSONRPCError

class Unauthenticated(JSONRPCError):
    code = -32010
    message = 'Authentication required'


app = Flask(__name__)

@dataclass
class FlaskContext:
    user_id: int | None
    ip_address: str

@dataclass
class SecureOperationParams:
    action: str

@dataclass
class SecureOperationResult:
    action: str
    executed_by: int
    from_ip: str

class SecureOperation(Method):
    def execute(self, params: SecureOperationParams, context: FlaskContext) -> SecureOperationResult:
        if not context.user_id:
            raise Unauthenticated()
        return SecureOperationResult(
            action=params.action,
            executed_by=context.user_id,
            from_ip=context.ip_address,
        )

rpc = JSONRPC(version='2.0', context_type=FlaskContext)
rpc.register('secure_op', SecureOperation())

@app.route('/rpc', methods=['POST'])
def handle_rpc():
    user_id = request.headers.get('X-User-ID')
    ctx = FlaskContext(
        user_id=int(user_id) if user_id else None,
        ip_address=request.remote_addr,
    )
    response = rpc.handle(request.data, context=ctx)
    return response, 200, {'Content-Type': 'application/json'}

HTTP Request:

curl -X POST http://localhost:5000/rpc \
  -H "Content-Type: application/json" \
  -H "X-User-ID: 123" \
  -d '{
    "jsonrpc": "2.0",
    "method": "secure_op",
    "params": {"action": "transfer_funds"},
    "id": 1
  }'

FastAPI Integration

fastapi_context.py
from fastapi import FastAPI, Request, Header, Response
from dataclasses import dataclass
from jsonrpc import JSONRPC, Method
from jsonrpc.errors import JSONRPCError

class Unauthenticated(JSONRPCError):
    code = -32010
    message = 'Authentication required'


app = FastAPI()

@dataclass
class FastAPIContext:
    user_id: int | None
    username: str | None

@dataclass
class WhoAmIResult:
    user_id: int
    username: str | None

class WhoAmI(Method):
    def execute(self, params: None, context: FastAPIContext) -> WhoAmIResult:
        if not context.user_id:
            raise Unauthenticated()
        return WhoAmIResult(user_id=context.user_id, username=context.username)

rpc = JSONRPC(version='2.0', context_type=FastAPIContext)
rpc.register('whoami', WhoAmI())

@app.post('/rpc')
async def handle_rpc(
    request: Request,
    x_user_id: int | None = Header(None),
    x_username: str | None = Header(None),
):
    ctx = FastAPIContext(user_id=x_user_id, username=x_username)
    body = await request.body()
    response = await rpc.handle_async(body, context=ctx)
    if response is None:  # notification: nothing to send back
        return Response(status_code=204)
    return Response(content=response, media_type='application/json')

Hierarchical Context

Context types can form an inheritance hierarchy. A method that requires a more specific context type can only be registered with an RPC that provides exactly that type (or a subclass of it).

hierarchical_context.py
from dataclasses import dataclass
from jsonrpc import JSONRPC, MethodGroup, Method

@dataclass
class BaseContext:
    request_id: str

@dataclass
class AuthContext(BaseContext):
    user_id: int

@dataclass
class AdminContext(AuthContext):
    is_admin: bool

@dataclass
class AdminResult:
    admin: bool

class AdminAction(Method):
    def execute(self, params: None, context: AdminContext) -> AdminResult:
        return AdminResult(admin=context.is_admin)

# Works: AdminContext is a subclass of AuthContext
rpc = JSONRPC(version='2.0', context_type=AdminContext)
group = MethodGroup()
group.register('action', AdminAction())
rpc.register('admin', group)

# Pass a full AdminContext at call time
ctx = AdminContext(request_id='req-1', user_id=7, is_admin=True)
resp = rpc.handle('{"jsonrpc":"2.0","method":"admin.action","params":null,"id":1}', context=ctx)

Key Points

  • Context from transport: Extract from HTTP headers, sessions, JWT tokens
  • Type validation: Context types validated at registration time
  • Hierarchical: Method context type must be in the same inheritance chain as the RPC context type
  • Security: Implement authentication and authorization in execute()
  • Framework integration: Works with Flask, FastAPI, Django, etc.

Fail-Fast

Registering a method whose context type has no relationship to the RPC's context type raises TypeError immediately — before any requests are served.

What's Next?

→ Groups - Organize methods hierarchically