Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
120 changes: 115 additions & 5 deletions sdk/typescript/_bundled_plugin/scripts/finding_preview.py
Original file line number Diff line number Diff line change
Expand Up @@ -140,6 +140,8 @@ def bounded_finding_details(value: Any) -> dict[str, Any]:
"severity",
"status",
"taxonomy",
"preventiveControls",
"remediationTests",
Comment thread
mldangelo-oai marked this conversation as resolved.
Comment thread
mldangelo-oai marked this conversation as resolved.
):
if key in value:
prepared[key] = (
Expand All @@ -148,8 +150,64 @@ def bounded_finding_details(value: Any) -> dict[str, Any]:
else value[key]
)

budget = [FINDING_DETAILS_PREVIEW_BYTES]
bounded = bounded_json_value(prepared, budget)
guidance = {
key: prepared[key]
for key in ("remediationTests", "preventiveControls")
if key in prepared and isinstance(prepared[key], list)
}
diagnostics = (
"rootCause",
"root_cause",
"validation",
"attackPath",
"codeEvidence",
"code_evidence",
)
core_keys = (
"writeup",
*diagnostics,
"confidence",
"detectedAt",
"identity",
"provenance",
"ruleId",
"severity",
"status",
"taxonomy",
"evidence",
"evidenceExcerpt",
)
core = {key: prepared[key] for key in core_keys if key in prepared}
extras = {
key: item
for key, item in prepared.items()
if key not in core and key not in guidance
}
complete_guidance = {key: items[:1] for key, items in guidance.items()}
minimum_guidance = {
key: [items[0][:1]] if items and isinstance(items[0], str) else []
for key, items in guidance.items()
}
projected_core = {}
for selected_guidance in (complete_guidance, minimum_guidance):
reserved = (
len(json.dumps(selected_guidance, separators=(",", ":")).encode("utf-8")) - 1
if selected_guidance
else 0
)
if reserved >= FINDING_DETAILS_PREVIEW_BYTES:
continue
projected_core = bounded_json_value(
core,
[FINDING_DETAILS_PREVIEW_BYTES - reserved],
)
if all(key in projected_core for key in core):
break
ordered_guidance = dict(sorted(guidance.items(), key=lambda entry: bool(entry[1])))
bounded = bounded_json_value(
{**projected_core, **ordered_guidance, **extras},
[FINDING_DETAILS_PREVIEW_BYTES],
)
return bounded if isinstance(bounded, dict) else {}


Expand Down Expand Up @@ -215,11 +273,20 @@ def bounded_json_value(value: Any, budget: list[int], *, depth: int = 0) -> Any:
if not consume_json_budget(budget, 2):
return []
result = []
for item in value[:20]:
for item in value:
Comment thread
mldangelo-oai marked this conversation as resolved.
remaining = budget[0]
separator = 0 if not result else 1
if not consume_json_budget(budget, separator):
break
result.append(bounded_json_value(item, budget, depth=depth + 1))
bounded_item = bounded_json_value(item, budget, depth=depth + 1)
size = len(json.dumps(bounded_item, separators=(",", ":")).encode("utf-8"))
if separator + size > remaining or (
isinstance(item, str) and item and bounded_item == ""
):
budget[0] = remaining
break
budget[0] = remaining - separator - size
result.append(bounded_item)
return result
if isinstance(value, dict):
if not consume_json_budget(budget, 2):
Expand All @@ -228,13 +295,56 @@ def bounded_json_value(value: Any, budget: list[int], *, depth: int = 0) -> Any:
for key, item in list(value.items())[:20]:
if budget[0] <= 0 or not isinstance(key, str):
break
remaining = budget[0]
separator = 0 if not result else 1
if not consume_json_budget(budget, separator):
budget[0] = remaining
break
bounded_key, key_size = bounded_json_text(key, min(budget[0], 512))
if not consume_json_budget(budget, key_size + 1):
budget[0] = remaining
break
item_budget = budget
if depth == 0 and key == "remediationTests":
controls = value.get("preventiveControls")
if (
isinstance(item, list)
and item
and isinstance(item[0], str)
and item[0]
and isinstance(controls, list)
and controls
and isinstance(controls[0], str)
and controls[0]
):
minimum_tests = len(
json.dumps([item[0][0]], separators=(",", ":")).encode("utf-8")
)
for control in (controls[0], controls[0][0]):
reserved = (
len(
json.dumps(
{"preventiveControls": [control]},
separators=(",", ":"),
).encode("utf-8")
)
- 1
)
if budget[0] >= minimum_tests + reserved:
item_budget = [budget[0] - reserved]
break
bounded_item = bounded_json_value(item, item_budget, depth=depth + 1)
size = (
separator
+ key_size
+ 1
+ len(json.dumps(bounded_item, separators=(",", ":")).encode("utf-8"))
)
if size > remaining or (isinstance(item, str) and item and bounded_item == ""):
budget[0] = remaining
break
result[bounded_key] = bounded_json_value(item, budget, depth=depth + 1)
budget[0] = remaining - size
result[bounded_key] = bounded_item
return result
consume_json_budget(budget, 4)
return None
Expand Down
154 changes: 154 additions & 0 deletions sdk/typescript/tests-ts/plugin-report-limits.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -39,4 +39,158 @@ describe("bundled scan report and source limits", () => {
unsafePathRejected: true,
});
});

test("preserves bounded remediation tests and preventive controls", () => {
const python = Bun.which("python3") ?? Bun.which("python");
expect(python).not.toBeNull();
const program = [
"import json, sys",
"sys.path.insert(0, sys.argv[1])",
"from finding_preview import bounded_finding_details",
"diagnostics = {'rootCause': {'summary': 'Missing authorization check.'}, 'validation': {'summary': 'An untrusted request reaches the protected resource.'}, 'attackPath': {'narrative': 'The request bypasses the authorization boundary.'}, 'codeEvidence': [{'id': 'evidence', 'label': 'Missing check', 'path': 'example.py', 'startLine': 1, 'code': 'return resource', 'explanation': 'No authorization check runs.'}], 'evidence': 'The protected resource was exposed.', 'evidenceExcerpt': 'return resource'}",
"details = {'remediationTests': [f'test-{index}' for index in range(40)], 'preventiveControls': [f'control-{index}' for index in range(40)]}",
"large = {**diagnostics, 'preventiveControls': ['x' * 900 for _ in range(20)], 'remediationTests': ['Verify authorization.'], 'writeup': {'reportPath': 'findings/example/example.md'}, 'provenance': {'source': 'scan'}, 'severity': {'level': 'high', 'rationale': 'Verified impact'}, 'status': 'open', 'taxonomy': {'category': 'injection', 'cwe': ['CWE-79']}}",
"code_evidence = [{'id': f'evidence-{index}', 'label': 'example', 'path': 'example.py', 'startLine': 1, 'code': 'c' * 1500, 'explanation': 'e' * 1500} for index in range(4)]",
"rich = {'rootCause': {'summary': 'r' * 2000}, 'validation': {'summary': 'v' * 3000}, 'attackPath': {'narrative': 'a' * 4000}, 'codeEvidence': code_evidence, 'evidenceExcerpt': 'e' * 8000, 'identity': {'anchor': 'finding'}, 'preventiveControls': ['Centralize authorization.'], 'remediationTests': ['Verify authorization.']}",
"boundary = {**diagnostics, 'remediationTests': ['x'] * 4000, 'preventiveControls': ['Keep authorization centralized.']}",
"unicode_boundary = {**diagnostics, 'remediationTests': ['😀'] * 2000, 'preventiveControls': ['🛡'] * 2000}",
"empty_controls = {**diagnostics, 'remediationTests': ['x'] * 4000, 'preventiveControls': []}",
"empty_tests = {**diagnostics, 'remediationTests': [], 'preventiveControls': ['control'] * 4000}",
"oversized_metadata = {**diagnostics, 'confidence': {'level': 'high', 'rationale': 'x' * 17000}, 'remediationTests': ['Verify authorization.'], 'preventiveControls': ['Centralize authorization.']}",
"oversized_guidance = {'rootCause': {'summary': 'root'}, 'validation': {'summary': 'validation'}, 'attackPath': {'narrative': 'attack'}, 'codeEvidence': [{'id': 'evidence', 'label': 'evidence', 'path': 'example.py', 'startLine': 1, 'code': 'x', 'explanation': 'evidence'}], 'evidence': 'legacy', 'evidenceExcerpt': 'excerpt', 'remediationTests': ['x' * 7800], 'preventiveControls': ['y' * 7930]}",
"nested_boundary = {'remediationTests': ['x'] * 3937, 'rootCause': {'summary': 'r' * 178, 'detail': {'x': {'y': 'z'}}}}",
"projections = {key: bounded_finding_details(value) for key, value in {'details': details, 'large': large, 'rich': rich, 'boundary': boundary, 'unicodeBoundary': unicode_boundary, 'emptyControls': empty_controls, 'emptyTests': empty_tests, 'oversizedMetadata': oversized_metadata, 'oversizedGuidance': oversized_guidance, 'nestedBoundary': nested_boundary}.items()}",
"print(json.dumps({'projections': projections, 'bytes': {key: len(json.dumps(value, separators=(',', ':')).encode()) for key, value in projections.items()}}))",
].join("\n");
const result = Bun.spawnSync(
[python!, "-I", "-B", "-c", program, join(PLUGIN_ROOT, "scripts")],
{ stdout: "pipe", stderr: "pipe" },
);

expect(result.exitCode, new TextDecoder().decode(result.stderr)).toBe(0);
const { projections, bytes } = JSON.parse(
new TextDecoder().decode(result.stdout),
) as {
projections: {
details: Record<string, unknown>;
large: Record<string, unknown>;
rich: Record<string, unknown>;
boundary: { remediationTests: string[]; preventiveControls: string[] };
unicodeBoundary: {
remediationTests: string[];
preventiveControls: string[];
};
emptyControls: {
remediationTests: string[];
preventiveControls: string[];
};
emptyTests: {
remediationTests: string[];
preventiveControls: string[];
};
oversizedMetadata: {
remediationTests: string[];
preventiveControls: string[];
};
oversizedGuidance: {
remediationTests: string[];
preventiveControls: string[];
};
nestedBoundary: {
remediationTests: string[];
rootCause: { summary: string };
};
};
bytes: Record<string, number>;
};
expect(projections.details).toEqual({
preventiveControls: Array.from(
{ length: 40 },
(_, index) => `control-${index}`,
),
remediationTests: Array.from(
{ length: 40 },
(_, index) => `test-${index}`,
),
});
expect(projections.large).toMatchObject({
writeup: { reportPath: "findings/example/example.md" },
provenance: { source: "scan" },
remediationTests: ["Verify authorization."],
severity: { level: "high", rationale: "Verified impact" },
status: "open",
taxonomy: { category: "injection", cwe: ["CWE-79"] },
});
expect(projections.rich).toMatchObject({
identity: { anchor: "finding" },
preventiveControls: ["Centralize authorization."],
remediationTests: ["Verify authorization."],
});
for (const finding of [
projections.large,
projections.rich,
projections.boundary,
projections.unicodeBoundary,
projections.emptyControls,
projections.emptyTests,
projections.oversizedMetadata,
projections.oversizedGuidance,
]) {
expect(finding).toMatchObject({
rootCause: { summary: expect.any(String) },
validation: { summary: expect.any(String) },
attackPath: { narrative: expect.any(String) },
codeEvidence: expect.arrayContaining([
expect.objectContaining({
id: expect.any(String),
path: "example.py",
}),
]),
});
}
for (const finding of [
projections.large,
projections.boundary,
projections.unicodeBoundary,
projections.emptyControls,
projections.emptyTests,
]) {
expect(finding).toMatchObject({
evidence: "The protected resource was exposed.",
evidenceExcerpt: "return resource",
});
}
expect(projections.rich).toHaveProperty("evidenceExcerpt");
expect(projections.oversizedGuidance).toMatchObject({
evidence: "legacy",
evidenceExcerpt: "excerpt",
});
expect(
projections.boundary.remediationTests.every((value) => value !== ""),
).toBe(true);
expect(projections.boundary.preventiveControls).toEqual([
"Keep authorization centralized.",
]);
expect(projections.unicodeBoundary.remediationTests.length).toBeGreaterThan(
0,
);
expect(
projections.unicodeBoundary.preventiveControls.every(
(value) => value === "🛡",
),
).toBe(true);
expect(
projections.unicodeBoundary.preventiveControls.length,
).toBeGreaterThan(0);
expect(projections.emptyControls.preventiveControls).toEqual([]);
expect(projections.emptyControls.remediationTests.length).toBeGreaterThan(
20,
);
expect(projections.emptyTests.remediationTests).toEqual([]);
expect(projections.emptyTests.preventiveControls.length).toBeGreaterThan(
20,
);
expect(projections.nestedBoundary.rootCause.summary).toContain("r");
expect(Object.values(bytes).every((value) => value <= 16_000)).toBe(true);
});
});
Loading