This is an automated email from the ASF dual-hosted git repository.
jason810496 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 259fe2d3e80 Warn on suspicious Dag and task IDs at TS SDK build time
(#70993)
259fe2d3e80 is described below
commit 259fe2d3e80f1ca4871f090fded5cd5025eb1cdf
Author: Andrew Chang <[email protected]>
AuthorDate: Thu Aug 20 07:29:54 2026 +0400
Warn on suspicious Dag and task IDs at TS SDK build time (#70993)
---
ts-sdk/src/cli/pack.ts | 8 +-
ts-sdk/src/cli/validate.ts | 66 ++++++++++++++
ts-sdk/src/coordinator/manifest.ts | 23 +----
ts-sdk/tests/cli/pack.test.ts | 46 +++++-----
ts-sdk/tests/cli/validate.test.ts | 101 ++++++++++++++++++++++
ts-sdk/tests/coordinator/runtime-manifest.test.ts | 50 ++++-------
6 files changed, 211 insertions(+), 83 deletions(-)
diff --git a/ts-sdk/src/cli/pack.ts b/ts-sdk/src/cli/pack.ts
index 3268ec9aab6..0642f0e13d3 100644
--- a/ts-sdk/src/cli/pack.ts
+++ b/ts-sdk/src/cli/pack.ts
@@ -21,9 +21,9 @@
// artifact NodeCoordinator consumes — `bundle.mjs` with the airflow
// metadata embedded as a leading `//# airflowMetadata=<base64>` comment.
//
-// Mirrors airflow-go-pack: build first, then run the built bundle with
-// --airflow-metadata so the manifest comes from the bundle's own task
-// registry and schema version, never from a hand-written sidecar.
+// Build first, then run the built bundle with --airflow-metadata so the
+// manifest comes from the bundle's own task registry and schema version,
+// never from a hand-written sidecar.
import { execFileSync } from "node:child_process";
import { readFileSync, rmSync, writeFileSync } from "node:fs";
@@ -34,6 +34,7 @@ import {
AIRFLOW_METADATA_SENTINEL,
type BundleManifest,
} from "../coordinator/manifest.js";
+import { warnOnSuspiciousIds } from "./validate.js";
const AIRFLOW_BUNDLE_METADATA_VERSION = "1.0";
const BUNDLE_FILENAME = "bundle.mjs";
@@ -243,6 +244,7 @@ export async function runPack(argv: readonly string[]):
Promise<void> {
process.stderr.write(`warning: dag ${JSON.stringify(dagId)} has no
tasks\n`);
}
}
+ warnOnSuspiciousIds(manifest.dags);
const metadataYaml = renderMetadataYaml({
airflow_bundle_metadata_version: AIRFLOW_BUNDLE_METADATA_VERSION,
diff --git a/ts-sdk/src/cli/validate.ts b/ts-sdk/src/cli/validate.ts
new file mode 100644
index 00000000000..91fc6e35482
--- /dev/null
+++ b/ts-sdk/src/cli/validate.ts
@@ -0,0 +1,66 @@
+/*!
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import type { BundleManifest } from "../coordinator/manifest.js";
+
+const MAX_ID_LENGTH = 250;
+const ID_REGEX = /^[\p{L}\p{N}_.-]+$/u;
+
+type WarnFn = (message: string) => void;
+
+/**
+ * Check every dag and task id in the manifest against the rules the Airflow
+ * server enforces (`airflow.utils.helpers.validate_key`). Best-effort and
+ * warn-only: the server validates authoritatively, and checks like the `..`
+ * one depend on server configuration the packer cannot see.
+ */
+export function warnOnSuspiciousIds(
+ dags: BundleManifest["dags"],
+ warn: WarnFn = (message) => process.stderr.write(`${message}\n`),
+): void {
+ for (const dagId of Object.keys(dags).sort()) {
+ warnOnSuspiciousId(`dag id ${JSON.stringify(dagId)}`, dagId, warn);
+ for (const taskId of dags[dagId]!.tasks) {
+ warnOnSuspiciousId(
+ `task id ${JSON.stringify(taskId)} in dag ${JSON.stringify(dagId)}`,
+ taskId,
+ warn,
+ );
+ }
+ }
+}
+
+function warnOnSuspiciousId(label: string, id: string, warn: WarnFn): void {
+ // Count code points, not UTF-16 units, to match the server-side len().
+ const length = [...id].length;
+ if (length > MAX_ID_LENGTH) {
+ warn(
+ `warning: ${label} is longer than ${MAX_ID_LENGTH} characters
(${length}); the Airflow server will reject it`,
+ );
+ }
+ if (!ID_REGEX.test(id)) {
+ warn(
+ `warning: ${label} must be made of alphanumeric characters, dashes,
dots, and underscores; the Airflow server will reject it`,
+ );
+ } else if (id.includes("..")) {
+ warn(
+ `warning: ${label} contains '..'; the Airflow server will reject it
unless [core] allow_double_dot_in_ids is enabled`,
+ );
+ }
+}
diff --git a/ts-sdk/src/coordinator/manifest.ts
b/ts-sdk/src/coordinator/manifest.ts
index 6baca7f4d7b..145cd350167 100644
--- a/ts-sdk/src/coordinator/manifest.ts
+++ b/ts-sdk/src/coordinator/manifest.ts
@@ -25,24 +25,6 @@ export const AIRFLOW_METADATA_FLAG = "--airflow-metadata";
/** Marks the manifest line on stdout, which import-time logging may also
reach. */
export const AIRFLOW_METADATA_SENTINEL = "__AIRFLOW_METADATA__ ";
-// Mirrors the Python task-SDK KEY_REGEX and validate_key in
airflow.sdk.definitions._internal.node.
-// Checked here rather than in Dag()/task() registration: that runs on every
-// bundle module load, including at task-execution-runtime startup, whereas
-// this only runs once, when the bundle is packed.
-const KEY_REGEX = /^[\p{L}\p{N}_.-]+$/u;
-const MAX_KEY_LENGTH = 250;
-
-function validateKey(label: string, value: string): void {
- if (typeof value !== "string" || !KEY_REGEX.test(value)) {
- throw new Error(
- `${label} must be made of alphanumeric characters, dashes, dots, and
underscores`,
- );
- }
- if (value.length > MAX_KEY_LENGTH) {
- throw new Error(`${label} must be less than ${MAX_KEY_LENGTH} characters,
not ${value.length}`);
- }
-}
-
/** Bundle manifest fields only the built bundle itself knows: the schema
* version it was compiled against and the Dag/task pairs it registered.
* Registered Dags without tasks appear with an empty `tasks` list so
@@ -56,9 +38,8 @@ export interface BundleManifest {
export function buildBundleManifest(registry: DagRegistry): BundleManifest {
const dags: BundleManifest["dags"] = {};
for (const { dagId, tasks } of listRegistryDags(registry)) {
- validateKey(`Dag "${dagId}"`, dagId);
- for (const taskId of tasks) {
- validateKey(`Task "${taskId}" of Dag "${dagId}"`, taskId);
+ if (typeof dagId !== "string") {
+ throw new Error("Dag ID must be a string");
}
Object.defineProperty(dags, dagId, {
configurable: true,
diff --git a/ts-sdk/tests/cli/pack.test.ts b/ts-sdk/tests/cli/pack.test.ts
index 6de652e7f28..75b2ccb5a45 100644
--- a/ts-sdk/tests/cli/pack.test.ts
+++ b/ts-sdk/tests/cli/pack.test.ts
@@ -211,38 +211,34 @@ describe("runPack", () => {
dagId: "bad id!",
taskId: "valid_task",
expected:
- 'Error: Dag "bad id!" must be made of alphanumeric characters, dashes,
dots, and underscores',
+ 'warning: dag id "bad id!" must be made of alphanumeric characters,
dashes, dots, and underscores; the Airflow server will reject it\n',
},
{
label: "task",
dagId: "valid_dag",
taskId: "bad id!",
expected:
- 'Error: Task "bad id!" of Dag "valid_dag" must be made of alphanumeric
characters, dashes, dots, and underscores',
+ 'warning: task id "bad id!" in dag "valid_dag" must be made of
alphanumeric characters, dashes, dots, and underscores; the Airflow server will
reject it\n',
},
- ])(
- "reports an invalid $label ID without a staging-bundle stack",
- async ({ dagId, taskId, expected }) => {
- outdir = mkdtempSync(path.join(tmpdir(), "ts-pack-"));
- const entry = path.join(outdir, "invalid-id-entry.ts");
- writeFileSync(
- entry,
- [
- `import { Dag, DagRegistry, serveDags } from
${JSON.stringify(SDK_INDEX)};`,
- `const invalidDag = new Dag(${JSON.stringify(dagId)});`,
- `invalidDag.task(${JSON.stringify(taskId)}, async () => undefined);`,
- "await serveDags(new DagRegistry(invalidDag));",
- ].join("\n"),
- );
-
- await expect(runPack([entry, "--outdir",
outdir])).rejects.toHaveProperty(
- "message",
- expected,
- );
- expect(existsSync(path.join(outdir, "bundle.mjs"))).toBe(false);
- expect(existsSync(path.join(outdir,
"bundle.pack-staging.mjs"))).toBe(false);
- },
- );
+ ])("warns on a suspicious $label ID but still packs", async ({ dagId,
taskId, expected }) => {
+ outdir = mkdtempSync(path.join(tmpdir(), "ts-pack-"));
+ const entry = path.join(outdir, "suspicious-id-entry.ts");
+ writeFileSync(
+ entry,
+ [
+ `import { Dag, DagRegistry, serveDags } from
${JSON.stringify(SDK_INDEX)};`,
+ `const suspiciousDag = new Dag(${JSON.stringify(dagId)});`,
+ `suspiciousDag.task(${JSON.stringify(taskId)}, async () =>
undefined);`,
+ "await serveDags(new DagRegistry(suspiciousDag));",
+ ].join("\n"),
+ );
+ const stderr = captureStderr();
+
+ await runPack([entry, "--outdir", outdir]);
+
+ expect(stderr()).toContain(expected);
+ expect(existsSync(path.join(outdir, "bundle.mjs"))).toBe(true);
+ });
it("reports the last error from a failed bundle", async () => {
outdir = mkdtempSync(path.join(tmpdir(), "ts-pack-"));
diff --git a/ts-sdk/tests/cli/validate.test.ts
b/ts-sdk/tests/cli/validate.test.ts
new file mode 100644
index 00000000000..f97dba81d4c
--- /dev/null
+++ b/ts-sdk/tests/cli/validate.test.ts
@@ -0,0 +1,101 @@
+/*!
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import { describe, expect, it } from "vitest";
+
+import { warnOnSuspiciousIds } from "../../src/cli/validate.js";
+import type { BundleManifest } from "../../src/coordinator/manifest.js";
+
+// U+20000, a CJK letter outside the Basic Multilingual Plane: one code point
+// but two UTF-16 units, so it separates code-point from .length counting.
+const ASTRAL_LETTER = "𠀀";
+
+function warningsFor(dags: BundleManifest["dags"]): string[] {
+ const warnings: string[] = [];
+ warnOnSuspiciousIds(dags, (message) => warnings.push(message));
+ return warnings;
+}
+
+function dagCharsetWarning(id: string): string {
+ return `warning: dag id ${JSON.stringify(id)} must be made of alphanumeric
characters, dashes, dots, and underscores; the Airflow server will reject it`;
+}
+
+describe("warnOnSuspiciousIds", () => {
+ it.each([
+ "simple",
+ "with-dash",
+ "with.dot",
+ "with_underscore",
+ "0numeric",
+ "café_dag",
+ "任務",
+ "a".repeat(250),
+ "任".repeat(250),
+ ASTRAL_LETTER.repeat(250),
+ ])("does not warn on valid id %j", (id) => {
+ expect(warningsFor({ [id]: { tasks: [id] } })).toEqual([]);
+ });
+
+ it.each(["a".repeat(251), "任".repeat(251), ASTRAL_LETTER.repeat(251)])(
+ "warns on an id longer than 250 code points: %j",
+ (id) => {
+ expect(warningsFor({ [id]: { tasks: [] } })).toEqual([
+ `warning: dag id ${JSON.stringify(id)} is longer than 250 characters
(251); the Airflow server will reject it`,
+ ]);
+ },
+ );
+
+ // "a..b c" also locks the else-if: a charset failure suppresses the '..'
warning.
+ it.each(["", "with space", "with/slash", "with:colon", "with\ttab", "a..b
c"])(
+ "warns on an id with invalid characters: %j",
+ (id) => {
+ expect(warningsFor({ [id]: { tasks: [] }
})).toEqual([dagCharsetWarning(id)]);
+ },
+ );
+
+ it("warns on an id containing double dots", () => {
+ expect(warningsFor({ "a..b": { tasks: [] } })).toEqual([
+ `warning: dag id "a..b" contains '..'; the Airflow server will reject it
unless [core] allow_double_dot_in_ids is enabled`,
+ ]);
+ });
+
+ it("warns twice on an id that is both too long and invalid", () => {
+ const id = "a".repeat(250) + " b";
+ expect(warningsFor({ [id]: { tasks: [] } })).toEqual([
+ `warning: dag id ${JSON.stringify(id)} is longer than 250 characters
(252); the Airflow server will reject it`,
+ dagCharsetWarning(id),
+ ]);
+ });
+
+ it("sorts dag ids for stable output", () => {
+ const warnings = warningsFor({
+ "delta d": { tasks: [] },
+ "alpha d": { tasks: [] },
+ "charlie d": { tasks: [] },
+ "bravo d": { tasks: [] },
+ });
+ expect(warnings).toEqual(["alpha d", "bravo d", "charlie d", "delta
d"].map(dagCharsetWarning));
+ });
+
+ it("names the owning dag in a task id warning", () => {
+ expect(warningsFor({ my_dag: { tasks: ["bad task"] } })).toEqual([
+ `warning: task id "bad task" in dag "my_dag" must be made of
alphanumeric characters, dashes, dots, and underscores; the Airflow server will
reject it`,
+ ]);
+ });
+});
diff --git a/ts-sdk/tests/coordinator/runtime-manifest.test.ts
b/ts-sdk/tests/coordinator/runtime-manifest.test.ts
index 9c8beec94e1..f1e4ea76b17 100644
--- a/ts-sdk/tests/coordinator/runtime-manifest.test.ts
+++ b/ts-sdk/tests/coordinator/runtime-manifest.test.ts
@@ -65,48 +65,30 @@ describe("buildBundleManifest", () => {
expect(Object.keys(buildBundleManifest(registry).dags)).toEqual(["dag_a"]);
});
- it("rejects an empty dagId", () => {
- const registry = new DagRegistry(buildDag(""));
- expect(() => buildBundleManifest(registry)).toThrowError(/must be made of
alphanumeric/);
- });
-
- it("rejects an empty taskId", () => {
- const registry = new DagRegistry(buildDag("example_dag", ""));
- expect(() => buildBundleManifest(registry)).toThrowError(/must be made of
alphanumeric/);
- });
-
- it.each([" ", "\t", "my dag", "a/b", "task@1"])(
- "rejects a dagId with characters no Python dag_id allows: %j",
+ // The server would reject these ids. The manifest keeps them and
+ // airflow-ts-pack warns at build time instead of failing the pack.
+ it.each(["", " ", "\t", "my dag", "a/b", "task@1", "d".repeat(251)])(
+ "keeps a dagId the server would reject visible in the manifest: %j",
(dagId) => {
- const registry = new DagRegistry(buildDag(dagId));
- expect(() => buildBundleManifest(registry)).toThrowError(/must be made
of alphanumeric/);
+ const manifest = buildBundleManifest(new DagRegistry(buildDag(dagId,
"t1")));
+ expect(manifest.dags[dagId]).toEqual({ tasks: ["t1"] });
},
);
- it.each([" ", "\t", "my task", "a/b", "task@1"])(
- "rejects a taskId with characters no Python task_id allows: %j",
+ // An empty taskId is absent: the bundle schema requires task ids to be
+ // non-empty strings, so airflow-ts-pack rejects it instead of warning.
+ it.each([" ", "\t", "my task", "a/b", "task@1", "t".repeat(251)])(
+ "keeps a taskId the server would reject visible in the manifest: %j",
(taskId) => {
- const registry = new DagRegistry(buildDag("example_dag", taskId));
- expect(() => buildBundleManifest(registry)).toThrowError(/must be made
of alphanumeric/);
+ const manifest = buildBundleManifest(new
DagRegistry(buildDag("example_dag", taskId)));
+ expect(manifest.dags["example_dag"]).toEqual({ tasks: [taskId] });
},
);
- it("rejects a dagId longer than 250 characters", () => {
- const registry = new DagRegistry(buildDag("d".repeat(251)));
- expect(() => buildBundleManifest(registry)).toThrowError(
- /must be less than 250 characters, not 251/,
- );
- });
-
- it("rejects a taskId longer than 250 characters", () => {
- const registry = new DagRegistry(buildDag("example_dag", "t".repeat(251)));
- expect(() => buildBundleManifest(registry)).toThrowError(
- /must be less than 250 characters, not 251/,
- );
- });
-
- it("does not validate key format when Dags are only registered, not packed",
() => {
- expect(() => new DagRegistry(buildDag("bad dag id", "bad task
id"))).not.toThrow();
+ it("rejects a non-string dagId before object-key coercion hides it", () => {
+ const dag = new Dag(123 as unknown as string);
+ dag.task("t1", async () => undefined);
+ expect(() => buildBundleManifest(new DagRegistry(dag))).toThrowError(/Dag
ID must be a string/);
});
});
