mirror of
https://github.com/morluto/rea
synced 2026-10-26 14:25:49 -04:00
morluto
c6ab965062
Keep signed co_stacksize and co_flags for historical magic entries whose constructors accepted them. Preserve magic 3410 with a constructor-revision limitation; reject negative values from 3411 onward. Fixes #6487
559 lines
18 KiB
TypeScript
559 lines
18 KiB
TypeScript
import { mkdir, symlink, writeFile } from "node:fs/promises";
|
|
import { join } from "node:path";
|
|
import { expect } from "vitest";
|
|
import { toolContract } from "../../src/contracts/toolContracts.js";
|
|
import {
|
|
marshalCode,
|
|
pythonPycFixture,
|
|
pythonPycHeader,
|
|
} from "../../src/python/PythonPyc.fixture.js";
|
|
import { createTestTempDirectory } from "../fixtures/temporaryDirectory.js";
|
|
import { cliTest } from "../support/cli/cliFixture.js";
|
|
|
|
const marshalUtf8 = (bytes: Buffer): Buffer => {
|
|
const length = Buffer.alloc(4);
|
|
length.writeUInt32LE(bytes.length);
|
|
return Buffer.concat([Buffer.from([0x75]), length, bytes]);
|
|
};
|
|
|
|
cliTest(
|
|
"keeps distinct Python code names distinct when JavaScript display strings collide",
|
|
async ({ cli }) => {
|
|
const root = await createTestTempDirectory("rea-python-pyc-unicode-name-");
|
|
const scalar = Buffer.from([0xf0, 0x9f, 0x98, 0x80]);
|
|
const surrogatePair = Buffer.from([0xed, 0xa0, 0xbd, 0xed, 0xb8, 0x80]);
|
|
const code = (name: Buffer) =>
|
|
marshalCode("py311", {
|
|
nameObject: marshalUtf8(name),
|
|
qualnameObject: marshalUtf8(name),
|
|
});
|
|
const frozenSet = Buffer.concat([
|
|
Buffer.from([0x3e, 2, 0, 0, 0]),
|
|
code(scalar),
|
|
code(surrogatePair),
|
|
]);
|
|
const path = join(root, "unicode-names.pyc");
|
|
await writeFile(
|
|
path,
|
|
Buffer.concat([
|
|
pythonPycHeader(3531),
|
|
marshalCode("py311", { consts: [frozenSet] }),
|
|
]),
|
|
);
|
|
|
|
const result = await cli.run({
|
|
arguments: ["inspect-python-pyc", path, "--json"],
|
|
});
|
|
expect(
|
|
result.exitCode,
|
|
`${result.stderr}\n${JSON.stringify(result.json)}`,
|
|
).toBe(0);
|
|
const evidence = toolContract("inspect_python_pyc").outputSchema.parse(
|
|
result.json,
|
|
);
|
|
const codes = evidence.normalized_result.code_objects;
|
|
expect(codes[1]?.name).toBe(codes[2]?.name);
|
|
expect(codes[0]?.consts).toEqual([
|
|
expect.objectContaining({ kind: "frozenset", length: 2 }),
|
|
]);
|
|
},
|
|
);
|
|
|
|
cliTest(
|
|
"reports nested CPython code objects through the public CLI",
|
|
async ({ cli }) => {
|
|
const root = await createTestTempDirectory("rea-python-pyc-cli-");
|
|
const path = join(root, "sample.pyc");
|
|
await writeFile(path, pythonPycFixture("py311", { nested: true }));
|
|
const result = await cli.run({
|
|
arguments: ["inspect-python-pyc", path, "--json"],
|
|
});
|
|
expect(result.exitCode, result.stderr).toBe(0);
|
|
const evidence = toolContract("inspect_python_pyc").outputSchema.parse(
|
|
result.json,
|
|
);
|
|
expect(evidence.normalized_result.header.python_version).toBe("3.12");
|
|
expect(evidence.normalized_result.code_objects[1]?.name).toBe("inner");
|
|
},
|
|
);
|
|
|
|
cliTest(
|
|
"classifies unsupported 3.7 alpha headers without labeling their marshal data malformed",
|
|
async ({ cli }) => {
|
|
const root = await createTestTempDirectory("rea-python-pyc-alpha-header-");
|
|
const marshal = Buffer.from(
|
|
"e3000000000000000000000000010000004000000073040000006400530029014ea900720100000072010000007201000000fa066d6f642e7079da083c6d6f64756c653e010000007300000000",
|
|
"hex",
|
|
);
|
|
expect(marshal).toHaveLength(77);
|
|
for (const magic of [3390, 3391]) {
|
|
const path = join(root, `${magic}.pyc`);
|
|
const header = Buffer.alloc(12);
|
|
header.writeUInt16LE(magic);
|
|
header[2] = 0x0d;
|
|
header[3] = 0x0a;
|
|
await writeFile(path, Buffer.concat([header, marshal]));
|
|
const result = await cli.run({
|
|
arguments: ["inspect-python-pyc", path, "--json"],
|
|
});
|
|
expect(result.exitCode).toBe(1);
|
|
expect(result.json).toMatchObject({
|
|
code: "unsupported_target",
|
|
category: "unsupported_target",
|
|
});
|
|
expect(JSON.stringify(result.json)).toContain(
|
|
"unsupported 12-byte CPython header",
|
|
);
|
|
expect(result.json).not.toHaveProperty("normalized_result");
|
|
}
|
|
|
|
for (const magic of [3392, 3394]) {
|
|
const path = join(root, `${magic}.pyc`);
|
|
await writeFile(path, Buffer.concat([pythonPycHeader(magic), marshal]));
|
|
const result = await cli.run({
|
|
arguments: ["inspect-python-pyc", path, "--json"],
|
|
});
|
|
expect(result.exitCode, result.stderr).toBe(0);
|
|
const evidence = toolContract("inspect_python_pyc").outputSchema.parse(
|
|
result.json,
|
|
);
|
|
expect(evidence.normalized_result.header.magic).toBe(magic);
|
|
expect(evidence.normalized_result.code_objects[0]?.bytecode.bytes).toBe(
|
|
4,
|
|
);
|
|
}
|
|
},
|
|
);
|
|
|
|
cliTest(
|
|
"preserves historical signed code fields and reports the ambiguous 3410 constructor revision",
|
|
async ({ cli }) => {
|
|
const root = await createTestTempDirectory("rea-python-pyc-signed-fields-");
|
|
for (const { name, magic, layout, fields, expected } of [
|
|
{
|
|
name: "zero",
|
|
magic: 3531,
|
|
layout: "py311",
|
|
fields: { flags: 0 },
|
|
expected: { stacksize: 1, flags: 0 },
|
|
},
|
|
{
|
|
name: "unknown-positive-flags",
|
|
magic: 3531,
|
|
layout: "py311",
|
|
fields: { flags: 0x40000000 },
|
|
expected: { stacksize: 1, flags: 0x40000000 },
|
|
},
|
|
{
|
|
name: "negative-stacksize-py37",
|
|
magic: 3394,
|
|
layout: "py37",
|
|
fields: { stacksize: 0xffffffff },
|
|
expected: { stacksize: -1, flags: 0 },
|
|
},
|
|
{
|
|
name: "negative-flags-py38-alpha",
|
|
magic: 3401,
|
|
layout: "py37",
|
|
fields: { flags: 0x80000000 },
|
|
expected: { stacksize: 1, flags: -0x80000000 },
|
|
},
|
|
{
|
|
name: "negative-stack-and-flags-py38-ambiguous",
|
|
magic: 3410,
|
|
layout: "py38",
|
|
fields: { stacksize: 0xffffffff, flags: 0x80000000 },
|
|
expected: { stacksize: -1, flags: -0x80000000 },
|
|
},
|
|
] as const) {
|
|
const path = join(root, `${name}.pyc`);
|
|
await writeFile(
|
|
path,
|
|
Buffer.concat([pythonPycHeader(magic), marshalCode(layout, fields)]),
|
|
);
|
|
const result = await cli.run({
|
|
arguments: ["inspect-python-pyc", path, "--json"],
|
|
});
|
|
expect(result.exitCode, result.stderr).toBe(0);
|
|
const evidence = toolContract("inspect_python_pyc").outputSchema.parse(
|
|
result.json,
|
|
);
|
|
expect(evidence.normalized_result.code_objects[0]).toMatchObject(
|
|
expected,
|
|
);
|
|
if (magic === 3410)
|
|
expect(evidence.normalized_result.limitations).toEqual(
|
|
expect.arrayContaining([
|
|
expect.stringContaining("Magic 3410 is shared"),
|
|
]),
|
|
);
|
|
}
|
|
},
|
|
);
|
|
|
|
cliTest(
|
|
"rejects negative code fields after the pinned constructor validation transition",
|
|
async ({ cli }) => {
|
|
const root = await createTestTempDirectory(
|
|
"rea-python-pyc-signed-rejection-",
|
|
);
|
|
for (const { name, magic, layout, fields, message } of [
|
|
{
|
|
name: "negative-stacksize-py38-b2",
|
|
magic: 3411,
|
|
layout: "py38",
|
|
fields: { stacksize: 0xffffffff },
|
|
message: "co_stacksize is negative",
|
|
},
|
|
{
|
|
name: "negative-flags-py38-b2",
|
|
magic: 3411,
|
|
layout: "py38",
|
|
fields: { flags: 0x80000000 },
|
|
message: "co_flags is negative",
|
|
},
|
|
{
|
|
name: "negative-stacksize-py311",
|
|
magic: 3531,
|
|
layout: "py311",
|
|
fields: { stacksize: 0xffffffff },
|
|
message: "co_stacksize is negative",
|
|
},
|
|
{
|
|
name: "negative-flags-py311",
|
|
magic: 3531,
|
|
layout: "py311",
|
|
fields: { flags: 0x80000000 },
|
|
message: "co_flags is negative",
|
|
},
|
|
] as const) {
|
|
const path = join(root, `${name}.pyc`);
|
|
await writeFile(
|
|
path,
|
|
Buffer.concat([pythonPycHeader(magic), marshalCode(layout, fields)]),
|
|
);
|
|
const result = await cli.run({
|
|
arguments: ["inspect-python-pyc", path, "--json"],
|
|
});
|
|
expect(result.exitCode).toBe(1);
|
|
expect(result.json).toMatchObject({
|
|
code: "invalid_request",
|
|
category: "invalid_input",
|
|
details: {
|
|
operation: "inspect_python_pyc",
|
|
issues: [
|
|
{
|
|
reason: "invalid_format",
|
|
message: expect.stringContaining(message),
|
|
},
|
|
],
|
|
},
|
|
});
|
|
expect(result.json).not.toHaveProperty("normalized_result");
|
|
}
|
|
},
|
|
);
|
|
|
|
cliTest(
|
|
"preserves negative CPython first line numbers through the public CLI",
|
|
async ({ cli }) => {
|
|
const root = await createTestTempDirectory("rea-python-pyc-lines-");
|
|
const path = join(root, "sample.pyc");
|
|
const nested = marshalCode("py311", {
|
|
name: "inner",
|
|
firstlineno: -7,
|
|
});
|
|
const bytes = Buffer.concat([
|
|
pythonPycHeader(3531),
|
|
marshalCode("py311", {
|
|
firstlineno: -1,
|
|
consts: [nested],
|
|
}),
|
|
]);
|
|
await writeFile(path, bytes);
|
|
const result = await cli.run({
|
|
arguments: ["inspect-python-pyc", path, "--json"],
|
|
});
|
|
expect(result.exitCode, result.stderr).toBe(0);
|
|
const evidence = toolContract("inspect_python_pyc").outputSchema.parse(
|
|
result.json,
|
|
);
|
|
expect(evidence.normalized_result.code_objects[0]?.firstlineno).toBe(-1);
|
|
expect(evidence.normalized_result.code_objects[0]?.consts).toContainEqual({
|
|
kind: "code",
|
|
name: "inner",
|
|
firstlineno: -7,
|
|
});
|
|
},
|
|
);
|
|
|
|
cliTest(
|
|
"matches CPython reference slots for singleton and reference tags",
|
|
async ({ cli }) => {
|
|
const root = await createTestTempDirectory("rea-python-pyc-ref-slots-");
|
|
const cases = [
|
|
{
|
|
name: "flagged-none",
|
|
consts: "2903cefa01787200000000",
|
|
expected: [
|
|
{ kind: "none" },
|
|
{ kind: "str", value: "x" },
|
|
{ kind: "str", value: "x" },
|
|
],
|
|
},
|
|
{
|
|
name: "flagged-ref",
|
|
consts: "2904fa0161f200000000fa01627201000000",
|
|
expected: [
|
|
{ kind: "str", value: "a" },
|
|
{ kind: "str", value: "a" },
|
|
{ kind: "str", value: "b" },
|
|
{ kind: "str", value: "b" },
|
|
],
|
|
},
|
|
];
|
|
for (const testCase of cases) {
|
|
const path = join(root, `${testCase.name}.pyc`);
|
|
await writeFile(
|
|
path,
|
|
Buffer.concat([
|
|
pythonPycHeader(3531),
|
|
marshalCode("py311", {
|
|
constsObject: Buffer.from(testCase.consts, "hex"),
|
|
}),
|
|
]),
|
|
);
|
|
const result = await cli.run({
|
|
arguments: ["inspect-python-pyc", path, "--json"],
|
|
});
|
|
expect(result.exitCode, result.stderr).toBe(0);
|
|
const evidence = toolContract("inspect_python_pyc").outputSchema.parse(
|
|
result.json,
|
|
);
|
|
expect(evidence.normalized_result.code_objects[0]?.consts).toMatchObject(
|
|
testCase.expected,
|
|
);
|
|
}
|
|
},
|
|
);
|
|
|
|
cliTest(
|
|
"counts code keys using the selected CPython equality rules",
|
|
async ({ cli }) => {
|
|
const root = await createTestTempDirectory("rea-python-pyc-equality-");
|
|
const cases = [
|
|
{ magic: 3413, layout: "py38" as const, lengths: [1, 1] },
|
|
{ magic: 3439, layout: "py38" as const, lengths: [1, 1] },
|
|
{ magic: 3531, layout: "py311" as const, lengths: [2, 2] },
|
|
];
|
|
for (const [index, testCase] of cases.entries()) {
|
|
const code = (linetable: Buffer) =>
|
|
marshalCode(testCase.layout, {
|
|
name: "f",
|
|
filename: "fixture.py",
|
|
firstlineno: 1,
|
|
linetable,
|
|
});
|
|
const frozenSet = Buffer.concat([
|
|
Buffer.from([0x3e]),
|
|
Buffer.from([2, 0, 0, 0]),
|
|
code(Buffer.from([2, 1])),
|
|
code(Buffer.from([2, 2])),
|
|
]);
|
|
const dict = Buffer.concat([
|
|
Buffer.from([0x7b]),
|
|
code(Buffer.from([2, 1])),
|
|
Buffer.from([0x4e]),
|
|
code(Buffer.from([2, 2])),
|
|
Buffer.from([0x54, 0x30]),
|
|
]);
|
|
const path = join(root, `case-${index}.pyc`);
|
|
await writeFile(
|
|
path,
|
|
Buffer.concat([
|
|
pythonPycHeader(testCase.magic),
|
|
marshalCode(testCase.layout, { consts: [frozenSet, dict] }),
|
|
]),
|
|
);
|
|
const result = await cli.run({
|
|
arguments: ["inspect-python-pyc", path, "--json"],
|
|
});
|
|
expect(result.exitCode, result.stderr).toBe(0);
|
|
const evidence = toolContract("inspect_python_pyc").outputSchema.parse(
|
|
result.json,
|
|
);
|
|
expect(
|
|
evidence.normalized_result.code_objects[0]?.consts.map((value) =>
|
|
value.kind === "frozenset" || value.kind === "dict"
|
|
? value.length
|
|
: null,
|
|
),
|
|
).toEqual(testCase.lengths);
|
|
}
|
|
},
|
|
);
|
|
|
|
cliTest(
|
|
"uses constructor-normalized legacy flags for code keys and preserves raw flags",
|
|
async ({ cli }) => {
|
|
const root = await createTestTempDirectory(
|
|
"rea-python-pyc-effective-flags-",
|
|
);
|
|
const cases = [
|
|
{
|
|
name: "py37-empty-tables",
|
|
magic: 3394,
|
|
layout: "py37" as const,
|
|
freevars: [],
|
|
cellvars: [],
|
|
lengths: [1, 1],
|
|
},
|
|
{
|
|
name: "py310-nonempty-tables",
|
|
magic: 3439,
|
|
layout: "py38" as const,
|
|
freevars: ["capture"],
|
|
cellvars: [],
|
|
lengths: [1, 1],
|
|
},
|
|
{
|
|
name: "py311-raw-flags",
|
|
magic: 3531,
|
|
layout: "py311" as const,
|
|
freevars: [],
|
|
cellvars: [],
|
|
lengths: [2, 2],
|
|
},
|
|
];
|
|
for (const testCase of cases) {
|
|
const code = (flags: number) =>
|
|
marshalCode(testCase.layout, {
|
|
name: "f",
|
|
filename: "fixture.py",
|
|
firstlineno: 1,
|
|
flags,
|
|
freevars: testCase.freevars,
|
|
cellvars: testCase.cellvars,
|
|
});
|
|
const first = code(0);
|
|
const second = code(0x40);
|
|
const frozenSet = Buffer.concat([
|
|
Buffer.from([0x3e]),
|
|
Buffer.from([2, 0, 0, 0]),
|
|
first,
|
|
second,
|
|
]);
|
|
const dict = Buffer.concat([
|
|
Buffer.from([0x7b]),
|
|
first,
|
|
Buffer.from([0x4e]),
|
|
second,
|
|
Buffer.from([0x54, 0x30]),
|
|
]);
|
|
const path = join(root, `${testCase.name}.pyc`);
|
|
await writeFile(
|
|
path,
|
|
Buffer.concat([
|
|
pythonPycHeader(testCase.magic),
|
|
marshalCode(testCase.layout, { consts: [frozenSet, dict] }),
|
|
]),
|
|
);
|
|
const result = await cli.run({
|
|
arguments: ["inspect-python-pyc", path, "--json"],
|
|
});
|
|
expect(result.exitCode, result.stderr).toBe(0);
|
|
const evidence = toolContract("inspect_python_pyc").outputSchema.parse(
|
|
result.json,
|
|
);
|
|
expect(
|
|
evidence.normalized_result.code_objects[0]?.consts.map((value) =>
|
|
value.kind === "frozenset" || value.kind === "dict"
|
|
? value.length
|
|
: null,
|
|
),
|
|
).toEqual(testCase.lengths);
|
|
expect(
|
|
evidence.normalized_result.code_objects
|
|
.filter((object) => object.name === "f")
|
|
.map((object) => object.flags),
|
|
).toContain(0);
|
|
|
|
const rawPath = join(root, `raw-${testCase.name}.pyc`);
|
|
await writeFile(
|
|
rawPath,
|
|
Buffer.concat([
|
|
pythonPycHeader(testCase.magic),
|
|
marshalCode(testCase.layout, {
|
|
name: "f",
|
|
flags: 0x40,
|
|
freevars: testCase.freevars,
|
|
cellvars: testCase.cellvars,
|
|
}),
|
|
]),
|
|
);
|
|
const rawResult = await cli.run({
|
|
arguments: ["inspect-python-pyc", rawPath, "--json"],
|
|
});
|
|
expect(rawResult.exitCode, rawResult.stderr).toBe(0);
|
|
const rawEvidence = toolContract("inspect_python_pyc").outputSchema.parse(
|
|
rawResult.json,
|
|
);
|
|
expect(rawEvidence.normalized_result.code_objects[0]?.flags).toBe(0x40);
|
|
}
|
|
},
|
|
);
|
|
|
|
cliTest(
|
|
"resolves a relative pyc path against the working directory",
|
|
async ({ cli }) => {
|
|
const root = await createTestTempDirectory("rea-python-pyc-cli-rel-");
|
|
await writeFile(join(root, "sample.pyc"), pythonPycFixture("py37"));
|
|
const result = await cli.run({
|
|
arguments: ["inspect-python-pyc", "sample.pyc", "--json"],
|
|
cwd: root,
|
|
});
|
|
expect(result.exitCode, result.stderr).toBe(0);
|
|
const evidence = toolContract("inspect_python_pyc").outputSchema.parse(
|
|
result.json,
|
|
);
|
|
expect(evidence.normalized_result.header.python_version).toBe("3.7");
|
|
expect(evidence.subject?.local_path).toBe(join(root, "sample.pyc"));
|
|
},
|
|
);
|
|
|
|
cliTest.skipIf(process.platform === "win32")(
|
|
"reads the file the symlink path names, not its lexical sibling",
|
|
async ({ cli }) => {
|
|
const root = await createTestTempDirectory("rea-python-pyc-cli-link-");
|
|
const target = join(root, "target");
|
|
await mkdir(join(target, "child"), { recursive: true });
|
|
await writeFile(join(target, "sample.pyc"), pythonPycFixture("py37"));
|
|
await writeFile(join(root, "sample.pyc"), pythonPycFixture("py311"));
|
|
await symlink(join(target, "child"), join(root, "link"));
|
|
const result = await cli.run({
|
|
arguments: ["inspect-python-pyc", "link/../sample.pyc", "--json"],
|
|
cwd: root,
|
|
});
|
|
expect(result.exitCode, result.stderr).toBe(0);
|
|
const evidence = toolContract("inspect_python_pyc").outputSchema.parse(
|
|
result.json,
|
|
);
|
|
expect(evidence.normalized_result.header.python_version).toBe("3.7");
|
|
expect(evidence.subject?.local_path).toBe(`${root}/link/../sample.pyc`);
|
|
},
|
|
);
|
|
|
|
cliTest("reports a truncated pyc as invalid input", async ({ cli }) => {
|
|
const root = await createTestTempDirectory("rea-python-pyc-cli-bad-");
|
|
const path = join(root, "broken.pyc");
|
|
await writeFile(path, Buffer.from("not a pyc"));
|
|
const result = await cli.run({
|
|
arguments: ["inspect-python-pyc", path, "--json"],
|
|
});
|
|
expect(result.exitCode).toBe(1);
|
|
expect(result.json).toMatchObject({
|
|
code: "invalid_request",
|
|
details: { operation: "inspect_python_pyc" },
|
|
});
|
|
});
|