Fix file upload security loopholes - #14

* Add file upload validation

* Code formatting

* Code formatting

* Change values to constants

…ranches, stale comments

Review findings on #2284, highest first.

#1 (High) — half the rolling-deploy shim had no gating test that runs in the
default lane. The create_workflow_execution response write had none at all, and
the PgBarrier descriptor was pinned only by a test behind the barrier_db fixture,
which skips wherever Postgres is absent. Descriptor construction moves to
build_callback_descriptor(), pinned by a DB-free suite
(workers/tests/test_legacy_transport_shim.py); the response write is pinned by a
new backend case. All four writes now fail a required build if dropped early.

#2 (High) — _process_file_batch_core's Args entries still described
barrier_context=None as the supported Celery chord path while the body treats it
as a malformed payload. Both entries rewritten; the log line now carries
execution/workflow ids and the batch size.

#3 — process_file_batch_django_compat delegates with two arguments, so after this
PR every call landed in that malformed-payload branch: a false ERROR, and a full
batch run with no claim (no pg_batch_dedup marker, so a redelivery re-runs it —
LLM spend twice and a duplicate destination write with use_file_history=False).
Its only producer was the Celery chord this PR removes, so it now refuses, the
same fail-fast choice step_execution makes. Its MRQ helpers are left in place per
the no-removal rule.

#5 — PG_TRANSPORT_CALLBACK_KWARG became unconditional but its consumers still
branched on it, leaving an else branch that swallows a failed finalization and
strands the execution. The marker is now popped for wire compatibility only;
the duplicate guard and the re-raise are unconditional. is_pg on
_update_execution_status_unified becomes raise_on_failure (default True); the one
caller inside an except block opts out so a raise cannot mask the original error.

#6 — the removal checklist now lists all five artefact groups, and
EXECUTION_EXCLUDED_PARAMS names the key through LEGACY_TRANSPORT_KEY instead of a
hardcoded literal grep would miss.

#7 — CallbackDescriptor.transport is Literal["pg_queue"] and required, not
NotRequired[str]: it is mandatory this release, "celery" must never be written,
and the follow-up removal becomes a type error at the write site.

#8 — the Barrier protocol's "two call sites still program against it" was untrue;
both construct PgBarrier concretely. Corrected, with what actually keeps it.

#10 — executor_dispatch_fakes reimplemented the queue-naming rule and accepted
any enqueue signature. queue_for now delegates to a new production helper
queue_for_executor() (also used by the three dispatch sites), and the fake mirrors
the QueueTransport protocol's keyword-only key set.

#11, #14 — stale "PG only" / "No-op on Celery" comments at the sites this PR made
unconditional, a citation of the deleted Celery max_retries=0 wrapper, and a
cross-reference to a chart file that lives in the cloud repo.

Five tests that asserted the removed Celery branches are restated against the
collapsed behaviour rather than deleted.

Verification: workers 1481 passed / 1 skipped; backend shim, dispatch-orchestrator
and async-wait tests pass; pre-commit clean.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

…, backend and SDK (#2284)

* UN-4078 [MISC] Delete the SDK Celery ExecutionDispatcher and retarget its tests

First slice of the Celery transport removal. `ExecutionDispatcher` published
`execute_extraction` to RabbitMQ and blocked on an `AsyncResult`; nothing has
selected it since UN-4046 made `get_executor_dispatcher()` return the PG
request-reply dispatcher unconditionally. It had zero non-test importers.

Deleted: `unstract/sdk1/.../execution/dispatcher.py` (316 lines), its export
from the package `__init__`, and the `TestExecutionDispatcher` suite.

The 11 workers test files that imported it are retargeted rather than dropped —
they encode live product behaviour (which queue each operation lands on, what
the payload carries), and that behaviour survives on `PgExecutionDispatcher`:

- Queue naming moves to `unstract.workflow_execution.executor_rpc.QUEUE_PREFIX`,
  the surviving definition. The `celery_executor_*` names are NOT dead Celery
  surface — `worker-pg-executor` subscribes to exactly those strings — so the
  assertions pin the literal wire names.
- New `tests/executor_dispatch_fakes.py` holds the shared fake transport, an
  eager variant that runs the task in-process (replacing the `send_task`
  monkey-patching four round-trip tests did), and a correctly-shaped callback
  signature builder.
- The header-forwarding suite is replaced, not ported: PG dispatch takes no
  `headers=` by design, so the new tests assert org routing rides the payload
  AND that passing `headers=` still raises — the exact mistake that broke three
  call sites during UN-4046.
- The raw-`send_task` canary is kept and its message updated; it matters more
  now, since a raw send_task publishes to a broker nothing drains.

Two docstrings that claimed executor calls "go through Celery" were corrected —
they became false with this change, not merely dated.

Verification: workers 1517 passed / 1 skipped. sdk1 unchanged at 20 pre-existing
failures (`test_llm_compat`, `test_litellm_cohere_timeout`), reproduced on a
stashed tree to confirm they predate this work. Pre-commit clean on all 17 files.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

* UN-4078 [MISC] Delete the Celery execution transport

Removes the Celery fan-out substrate and the per-execution transport field.
Nothing has selected either since UN-4046 made `select_backend()` return PG
unconditionally, but both were still compiled, imported and shipped — and
importable Celery surface has already cost two incidents (UN-3779's hardcoded
dispatcher, and the `headers=` breakage at three call sites during UN-4046).

Deleted outright:
- `queue_backend/redis_barrier.py` (761 lines) — the DECR-counter substrate
- `CeleryChordBarrier` + `BarrierBackend` + `get_barrier()` + the
  `WORKER_BARRIER_BACKEND` env selector
- `WorkflowTransport`, `DEFAULT_WORKFLOW_TRANSPORT`, `normalize_transport`,
  `is_pg_transport` (88 lines of `unstract/core/data_models.py`)
- `barrier_pg_decr_and_check` — a thin @worker_task wrapper reachable only as a
  Celery `.link`; the decrement runs in-body via `_barrier_pg_decrement`

Collapsed: the 10 `is_pg_transport` branch sites, the `transport` parameter
threaded through the general/api/scheduler orchestrators, the `transport` field
on `WorkflowContextData` and `FileProcessingContext`, the `is_pg` flag on the
file-batch path, and the `transport` key the backend wrote into the dispatch
payload. `_barrier_for_transport` becomes `PgBarrier()`.

Two guards become unconditional rather than PG-gated: the terminal-execution
skip and the duplicate-destination-write check. Both were only ever no-ops on
Celery.

This also starts retiring a documented live hazard. `DEFAULT_WORKFLOW_TRANSPORT` was
`"celery"`, so a payload that lost the field during a rolling deploy selected a
Celery fan-out with no consumers. Worth correcting the in-code note while
deleting it: it claimed `CeleryChordBarrier` was selected and no PG rows were
written, so the reaper could not see the strand. Neither held — every PG worker
sets `WORKER_BARRIER_BACKEND=pg`, so `PgBarrier` was selected, and its
`_reset_barrier` UPSERT is unconditional, so the barrier row was written and the
reaper did see it. Real behaviour was a ~2.5h delayed ERROR.

The reads are removed here, but the writes are NOT: a pre-UN-4078 worker still
defaults an absent field to Celery, so during a rolling deploy it would publish to
a RabbitMQ with no consumers. The next commit keeps writing the field for one
release; the hazard is gone only once that shim is removed.

`EXECUTION_EXCLUDED_PARAMS` deliberately keeps its `"transport"` entry: a
rolling deploy can still have an un-upgraded producer emitting the field, and
without the exclusion it reaches the legacy `execute_workflow` signature and
raises.

Tests: four whole suites deleted with their substrate (redis barrier, barrier
backend selection, barrier differential, workflow-context transport). The rest
are retargeted rather than dropped — the chord canary now asserts *zero* chord
call sites (with a non-vacuity lock, since a zero-expectation assertion is
exactly what a path bug passes silently), and the Celery-gated guard tests are
replaced by the unconditional behaviour they now have.

Verification: workers 1438 passed / 1 skipped / 0 failed. Pre-commit clean.
The backend suite cannot start in this environment (`AppRegistryNotReady`,
reproduced on a stashed tree — pre-existing), so the four backend edits are
CI-verified only.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

* UN-4078 [MISC] Keep writing the transport field for one release as a rolling-deploy shim

The previous commit removed every read of the `transport` payload field and
every write. Removing the writes in the same release is unsafe: a pre-UN-4078
worker still treats an absent field as "celery" (DEFAULT_WORKFLOW_TRANSPORT)
and, during a Kubernetes RollingUpdate, can receive a payload from an upgraded
producer. It then publishes to RabbitMQ, which has no Celery consumers, so:

- general workflows are marked ERROR by the reaper after ~2.5h, even with every
  file processed;
- API deployments take no orchestration claim, so nothing sweeps them and they
  stay EXECUTING.

The window is real: old file-processing pods keep draining claimed batches for
their full grace period (up to 9120s).

The four writes are restored as `transport: "pg_queue"`, all via one constant,
LEGACY_TRANSPORT_KEY / LEGACY_TRANSPORT_VALUE in unstract.core.data_models, so
the follow-up removal is a single grep:

- backend WorkflowHelper orchestrator dispatch payload
- backend create_workflow_execution response
- workers scheduler async_execute_bin kwargs
- workers PgBarrier callback descriptor (CallbackDescriptor gains
  `transport: NotRequired[str]`)

Nothing on the new side reads the field. Tests pin each write so the shim cannot
be dropped early; the scheduler cases also assert a stale "celery" in the
backend response is never forwarded.

Remove in the release after UN-4078: the writes, the constants, the
`transport` entry in EXECUTION_EXCLUDED_PARAMS, and test_legacy_transport_shim.py.

Verification: workers 1478 passed / 1 skipped / 0 failed; backend dispatch
tests 5 passed (with backend/sample.env loaded); pre-commit clean.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

* UN-4078 [MISC] Address standardized review: shim gating tests, dead branches, stale comments

Review findings on #2284, highest first.

#1 (High) — half the rolling-deploy shim had no gating test that runs in the
default lane. The create_workflow_execution response write had none at all, and
the PgBarrier descriptor was pinned only by a test behind the barrier_db fixture,
which skips wherever Postgres is absent. Descriptor construction moves to
build_callback_descriptor(), pinned by a DB-free suite
(workers/tests/test_legacy_transport_shim.py); the response write is pinned by a
new backend case. All four writes now fail a required build if dropped early.

#2 (High) — _process_file_batch_core's Args entries still described
barrier_context=None as the supported Celery chord path while the body treats it
as a malformed payload. Both entries rewritten; the log line now carries
execution/workflow ids and the batch size.

#3 — process_file_batch_django_compat delegates with two arguments, so after this
PR every call landed in that malformed-payload branch: a false ERROR, and a full
batch run with no claim (no pg_batch_dedup marker, so a redelivery re-runs it —
LLM spend twice and a duplicate destination write with use_file_history=False).
Its only producer was the Celery chord this PR removes, so it now refuses, the
same fail-fast choice step_execution makes. Its MRQ helpers are left in place per
the no-removal rule.

#5 — PG_TRANSPORT_CALLBACK_KWARG became unconditional but its consumers still
branched on it, leaving an else branch that swallows a failed finalization and
strands the execution. The marker is now popped for wire compatibility only;
the duplicate guard and the re-raise are unconditional. is_pg on
_update_execution_status_unified becomes raise_on_failure (default True); the one
caller inside an except block opts out so a raise cannot mask the original error.

#6 — the removal checklist now lists all five artefact groups, and
EXECUTION_EXCLUDED_PARAMS names the key through LEGACY_TRANSPORT_KEY instead of a
hardcoded literal grep would miss.

#7 — CallbackDescriptor.transport is Literal["pg_queue"] and required, not
NotRequired[str]: it is mandatory this release, "celery" must never be written,
and the follow-up removal becomes a type error at the write site.

#8 — the Barrier protocol's "two call sites still program against it" was untrue;
both construct PgBarrier concretely. Corrected, with what actually keeps it.

#10 — executor_dispatch_fakes reimplemented the queue-naming rule and accepted
any enqueue signature. queue_for now delegates to a new production helper
queue_for_executor() (also used by the three dispatch sites), and the fake mirrors
the QueueTransport protocol's keyword-only key set.

#11, #14 — stale "PG only" / "No-op on Celery" comments at the sites this PR made
unconditional, a citation of the deleted Celery max_retries=0 wrapper, and a
cross-reference to a chart file that lives in the cloud repo.

Five tests that asserted the removed Celery branches are restated against the
collapsed behaviour rather than deleted.

Verification: workers 1481 passed / 1 skipped; backend shim, dispatch-orchestrator
and async-wait tests pass; pre-commit clean.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5 <noreply@anthropic.com>