Introduction
In a typical API request, authorization happens before the application performs an operation.
A user sends a request, the backend authenticates the user, checks their permissions, and either allows or rejects the operation.
That approach works well for synchronous requests.
But what happens when the operation is not executed immediately?
Consider an application that allows administrators to schedule bulk device configuration updates. An operator has permission to update devices at a particular site. They submit a job, the API validates their permissions, and the job is placed in RabbitMQ.
Twenty minutes later, a background worker picks up the job.
However, during those twenty minutes, an administrator has revoked the operator's access to that site.
Should the worker still execute the operation?
This is an important security question in systems that use background workers, message queues, scheduled tasks, and asynchronous processing.
Authorization at the time of submission does not automatically mean authorization at the time of execution.
Let's understand why this happens and how to design a safer execution model using Python, FastAPI, and background workers.
1. The Authorization Gap in Asynchronous Systems
In a synchronous application, the request and operation usually happen within the same execution flow.
For example:
@app.post("/devices/{device_id}/configuration")
async def update_configuration(
device_id: str,
user: CurrentUser
):
await authorization.require(
user_id=user.id,
action="device:update",
resource_id=device_id
)
return await device_service.update(device_id)
The application checks authorization immediately before performing the operation.
Now imagine moving the operation into a background queue.
Client
|
v
FastAPI API
|
|-- Authenticate user
|-- Check permissions
|
v
RabbitMQ
|
| Time passes...
|
v
Background Worker
|
v
Device Configuration Update
The API has already completed its authorization check. The worker receives a message and executes it independently.
The problem is that permissions are not necessarily permanent.
Between job submission and execution:
An administrator may revoke the user's access.
The user's account may become inactive.
The resource may be moved to another tenant or site.
The user's role may change.
The requested operation may no longer be permitted under the current policy.
The queue does not automatically know about these changes.
A message broker transports messages. It does not enforce your application's authorization policies.
This creates an authorization gap between two important moments:
Submission time: When the user requests an operation.
Execution time: When the worker actually performs it.
A secure design must decide what authorization means at both stages.
2. Why Storing Permissions Inside a Job Payload Is Risky
One tempting solution is to include the user's permissions in the queued message.
For example:
job = {
"actor_id": user.id,
"device_id": device_id,
"permissions": [
"device:read",
"device:update"
]
}
The worker can then inspect the permissions and decide whether to continue.
This looks convenient, but it introduces a serious problem: the permissions represent an earlier state of the system.
Suppose the operator had device:update permission when the job was created.
The permission is revoked five minutes later.
The worker eventually processes the message and still sees:
"permissions": ["device:read", "device:update"]
The worker has no reliable way to know that the permission has changed.
This is a stale authorization decision.
A better approach is to store the identity and context required to evaluate authorization again, rather than treating serialized permissions as permanent approval.
For example:
job = {
"job_id": "job-123",
"actor_id": "user-456",
"tenant_id": "tenant-001",
"device_id": "device-789",
"action": "device:update"
}
The worker can use these identifiers to retrieve current information and evaluate the current authorization policy.
There is an important distinction here:
These responsibilities should not be mixed.
3. Implementing Execution-Time Authorization with FastAPI
Let's build a simplified example.
Assume an IoT platform where users can update device configurations through a FastAPI backend.
The API validates the request and submits a job to a queue. A separate worker processes the job.
Step 1: Validate permissions when submitting the job
The first authorization check should happen when the request is received.
This prevents unauthorized users from creating jobs in the first place.
from uuid import uuid4
async def queue_device_update(
user,
device_id: str,
queue
):
await authorization.require(
user_id=user.id,
action="device:update",
resource_id=device_id
)
job = {
"job_id": str(uuid4()),
"actor_id": user.id,
"tenant_id": user.tenant_id,
"device_id": device_id,
"action": "device:update"
}
await queue.publish(job)
return {
"message": "Device update queued",
"job_id": job["job_id"]
}
At this stage, the application confirms that the user is allowed to request the operation.
However, it does not assume that the same permission will remain valid indefinitely.
Step 2: Recheck permissions inside the worker
The worker should retrieve the initiating user's current state and evaluate the authorization policy again.
async def execute_device_update(job: dict):
actor = await users.get(job["actor_id"])
if not actor or not actor.is_active:
await audit.record(
event="BACKGROUND_JOB_DENIED",
actor_id=job["actor_id"],
resource_id=job["device_id"],
job_id=job["job_id"],
reason="Inactive or missing user"
)
return
allowed = await authorization.check(
user_id=actor.id,
tenant_id=job["tenant_id"],
action=job["action"],
resource_id=job["device_id"]
)
if not allowed:
await audit.record(
event="BACKGROUND_JOB_DENIED",
actor_id=actor.id,
resource_id=job["device_id"],
job_id=job["job_id"],
reason="Permission revoked"
)
return
await device_service.update_configuration(
job["device_id"]
)
The worker now evaluates authorization using current application state.
If the user's access has been revoked, the operation is rejected.
Notice that the worker does not trust a list of permissions received from the queue.
It uses the actor's identity and the authorization service to make a fresh decision.
Step 3: Keep authorization logic centralized
One mistake in larger applications is implementing permission checks independently in every API endpoint and worker.
Over time, these checks can become inconsistent.
A centralized authorization service provides a single place to evaluate access.
For example:
class AuthorizationService:
async def check(
self,
user_id: str,
tenant_id: str,
action: str,
resource_id: str
) -> bool:
user = await users.get(user_id)
if not user or not user.is_active:
return False
resource = await resources.get(resource_id)
if not resource:
return False
if resource.tenant_id != tenant_id:
return False
if user.tenant_id != tenant_id:
return False
permissions = await self.get_current_permissions(
user_id=user_id,
resource_id=resource_id
)
return action in permissions
This example uses a simplified permission lookup. A real implementation may use RBAC, ABAC, resource-level permissions, ownership rules, or a combination of them.
The important design principle is that both synchronous API requests and asynchronous workers should use the same authorization policy.
Otherwise, the API may reject an operation that the worker accidentally allows, or vice versa.
4. What Happens When Permissions Change During Execution?
Rechecking permissions when a job starts solves only part of the problem.
Consider a bulk operation involving 500 devices.
The worker checks authorization successfully and starts processing.
After 100 devices have been updated, an administrator revokes the user's access.
Should the remaining 400 devices still be updated?
There is no universal answer. It depends on the application's security requirements and business rules.
However, this decision should be explicit.
Option A: Authorization at job start
The worker checks permissions once before processing the job.
This is appropriate for some low-risk operations where the application intentionally treats authorization at execution start as sufficient.
The limitation is that permission changes during execution will not stop the remaining work.
Option B: Authorization at checkpoints
For sensitive or long-running operations, the worker periodically checks authorization.
async def execute_bulk_update(job):
for device_id in job["device_ids"]:
allowed = await authorization.check(
user_id=job["actor_id"],
tenant_id=job["tenant_id"],
action="device:update",
resource_id=device_id
)
if not allowed:
await jobs.mark_cancelled(
job["job_id"],
reason="Authorization revoked"
)
return
await device_service.update_configuration(
device_id
)
This provides more granular control.
However, checking every device can increase database and authorization-service traffic. Depending on the operation, checks may be grouped or performed at carefully selected checkpoints.
Option C: Explicit approval snapshots
Some business workflows intentionally preserve an approval decision.
For example, an authorized manager approves a financial transaction. The transaction may need to retain the approval even if the manager's role changes afterward.
In such cases, the application should store an explicit, auditable approval record with the relevant policy and approval context.
That is different from accidentally trusting stale permissions embedded in a queue message.
The application must distinguish between:
A job that requires current authorization at execution.
A job that has a valid, previously recorded business approval.
These are different security models.
5. Building an Auditable Authorization Lifecycle
Authorization decisions should be observable, particularly when operations happen outside the normal HTTP request lifecycle.
If a worker rejects a job, developers should be able to understand why.
A useful audit record might contain:
{
"event": "BACKGROUND_JOB_DENIED",
"job_id": "job-123",
"actor_id": "user-456",
"tenant_id": "tenant-001",
"resource_id": "device-789",
"action": "device:update",
"decision": "DENY",
"reason": "Permission revoked",
"execution_time": "2026-11-13T07:30:00Z"
}
For production systems, consider recording:
The initiating user or service identity.
The job ID and resource identifiers.
The requested action.
The authorization decision.
The reason for denial.
The execution timestamp.
The policy or authorization version, where applicable.
Whether the operation was started, completed, rejected, or cancelled.
Do not log passwords, access tokens, or other secrets.
It is also important to distinguish the initiating actor from the worker's technical identity.
The worker may authenticate to internal services using its own service credentials, but that does not mean it should lose the context of the user who initiated the operation.
A useful audit model records both:
Initiating actor: The user or service that requested the operation.
Executing identity: The background worker or service account that performed it.
This makes it easier to investigate unexpected operations and establish accountability.
6. Production Considerations Beyond Authorization
Rechecking authorization is important, but it should not be the only safeguard in a distributed job-processing system.
There are several related engineering concerns.
Idempotency
Message queues may deliver a message more than once.
A worker should not perform the same consequential operation repeatedly just because a message was redelivered.
Use a stable job or operation ID and an idempotency mechanism.
Resource ownership
Do not trust the tenant ID or resource ID in a message without validating their relationship.
The authorization service should confirm that the resource still belongs to the expected tenant and that the actor is allowed to access it.
Job state management
Track job states explicitly.
For example:
QUEUED
|
v
RUNNING
|
+----> COMPLETED
|
+----> FAILED
|
+----> DENIED
|
+----> CANCELLED
A denied job is not the same as a failed job. A cancelled job is not necessarily an execution error.
Keeping these states separate makes monitoring and troubleshooting more meaningful.
Authorization and side-effect boundaries
A permission check followed by a separate database or device operation can still have a race condition: permissions may change between the check and the actual side effect.
For high-risk operations, consider stronger coordination mechanisms, such as transactional checks where possible, authorization versioning, short-lived execution grants, or checks immediately before consequential actions.
No single approach eliminates every race across independent services. The correct design depends on the consistency guarantees and risk level of the operation.
Conclusion
Background jobs introduce a security boundary that is easy to overlook.
A request may be authorized when it enters the system, but the worker may execute it much later under a completely different authorization context.
The safest general approach is to separate job submission from job execution:
Authenticate and authorize the request when the job is created.
Store the initiating identity and the operation context.
Reevaluate current permissions when the worker executes the job.
Define how permission changes affect long-running operations.
Record authorization decisions in an auditable lifecycle.
For sensitive operations, authorization should also be considered at the point where consequential side effects occur.
The key takeaway: A queued message is an instruction to process work, not permanent proof that the work is authorized.
As applications become more asynchronous and distributed, authorization must be designed around the entire lifecycle of an operation—not just the initial API request.
Join the conversation! Your thoughts help the community grow.