mirror of
https://github.com/morluto/rea.git
synced 2026-10-26 17:35:11 +00:00
N0zoM1z0
b225b0f64b
Correlate typed labels with recovered block entries and one indirect dispatch. Preserve ambiguous and nonexact integer associations as unknown with raw diagnostics. Expose defaults separately, collect backing load metadata, and invalidate old profiles. Add independent compiler-byte CLI/MCP acceptance and actual Ghidra model controls. Assisted-by: Kogasa (gensokyo-skills:kogasa-surprise-testing) Assisted-by: Suika (gensokyo-skills:suika-scatter-gather-planning)
404 lines
13 KiB
TypeScript
404 lines
13 KiB
TypeScript
import { z } from "zod";
|
|
import { nativeDataTypeSchema } from "../domain/nativeDataType.js";
|
|
import {
|
|
nativeInstructionSchema,
|
|
nativeCallTargetsSchema,
|
|
} from "../domain/nativeInstruction.js";
|
|
|
|
import { AnalysisInputError, AnalysisOutputError } from "../domain/errors.js";
|
|
import {
|
|
functionBodyEntryAgrees,
|
|
functionInstructionWindowSchema,
|
|
functionDossierSchema,
|
|
type FunctionDossier,
|
|
} from "../domain/hopperValues.js";
|
|
import { jsonObjectSchema, type JsonValue } from "../domain/jsonValue.js";
|
|
import { nativeApiBoundarySchema } from "../domain/nativeApiBoundary.js";
|
|
import { nativeValueFlowSchema } from "../domain/nativeValueFlow.js";
|
|
import { err, ok, type Result } from "../domain/result.js";
|
|
import {
|
|
ghidraIdentifierSchema,
|
|
ghidraCanonicalAddressSchema,
|
|
ghidraFunctionClassificationSchema as classification,
|
|
ghidraFunctionBodySchema as ghidraFunctionBody,
|
|
ghidraProcedureIdentitySchema as procedureIdentity,
|
|
} from "./GhidraInventoryValues.js";
|
|
|
|
/** Read-only function operations admitted by the Ghidra adapter. */
|
|
export const GHIDRA_FUNCTION_OPERATIONS = [
|
|
"inspect_native_data_type",
|
|
"inspect_native_instruction",
|
|
"resolve_native_call_targets",
|
|
"analyze_function",
|
|
"procedure_assembly",
|
|
"procedure_callees",
|
|
"procedure_callers",
|
|
"procedure_info",
|
|
"procedure_pseudo_code",
|
|
"read_function_instructions",
|
|
"procedure_references",
|
|
"xrefs",
|
|
] as const;
|
|
|
|
/** One function-analysis operation implemented by the packaged Java bridge. */
|
|
export type GhidraFunctionOperation =
|
|
(typeof GHIDRA_FUNCTION_OPERATIONS)[number];
|
|
|
|
const operationSet: ReadonlySet<string> = new Set(GHIDRA_FUNCTION_OPERATIONS);
|
|
|
|
/** Narrow an application operation to the Ghidra function-analysis surface. */
|
|
export const isGhidraFunctionOperation = (
|
|
operation: string,
|
|
): operation is GhidraFunctionOperation => operationSet.has(operation);
|
|
|
|
const document = ghidraIdentifierSchema.nullable().default(null);
|
|
const procedure = ghidraIdentifierSchema;
|
|
const directProcedure = { document, procedure };
|
|
|
|
const inputSchemas = {
|
|
inspect_native_data_type: z
|
|
.object({
|
|
document,
|
|
type: z.string().min(1).nullable().default(null),
|
|
address: ghidraCanonicalAddressSchema.nullable().default(null),
|
|
})
|
|
.strict()
|
|
.refine(
|
|
(value) => (value.type === null) !== (value.address === null),
|
|
"Supply exactly one of type or address",
|
|
),
|
|
inspect_native_instruction: z
|
|
.object({ document, address: ghidraCanonicalAddressSchema })
|
|
.strict(),
|
|
resolve_native_call_targets: z
|
|
.object({ document, address: ghidraCanonicalAddressSchema })
|
|
.strict(),
|
|
analyze_function: z.object({ procedure }).strict(),
|
|
procedure_assembly: z.object(directProcedure).strict(),
|
|
procedure_callees: z.object(directProcedure).strict(),
|
|
procedure_callers: z.object(directProcedure).strict(),
|
|
procedure_info: z.object(directProcedure).strict(),
|
|
procedure_pseudo_code: z.object(directProcedure).strict(),
|
|
read_function_instructions: z.object(directProcedure).strict(),
|
|
procedure_references: z
|
|
.object({
|
|
...directProcedure,
|
|
direction: z.enum(["incoming", "outgoing"]).default("outgoing"),
|
|
})
|
|
.strict(),
|
|
xrefs: z.object({ document, address: ghidraCanonicalAddressSchema }).strict(),
|
|
} satisfies Readonly<Record<GhidraFunctionOperation, z.ZodType>>;
|
|
|
|
/** Validate and default one function request before it crosses the socket. */
|
|
export const parseGhidraFunctionInput = (
|
|
operation: GhidraFunctionOperation,
|
|
value: Readonly<Record<string, JsonValue>>,
|
|
): Result<Readonly<Record<string, JsonValue>>, AnalysisInputError> => {
|
|
const parsed = inputSchemas[operation].safeParse(value);
|
|
return parsed.success
|
|
? ok(
|
|
jsonObjectSchema.parse(
|
|
Object.fromEntries(
|
|
Object.entries(parsed.data).filter(
|
|
([, value]) => value !== undefined,
|
|
),
|
|
),
|
|
),
|
|
)
|
|
: err(new AnalysisInputError(operation, { cause: parsed.error }));
|
|
};
|
|
|
|
const localVariable = z
|
|
.object({
|
|
description: z.string(),
|
|
provenance: z.literal("ghidra-function-database"),
|
|
})
|
|
.strict();
|
|
const referenceKind = z
|
|
.object({
|
|
available: z.literal(true),
|
|
provenance: z.literal("ghidra-reference-manager"),
|
|
type: z.string().min(1),
|
|
flow: z.boolean(),
|
|
call: z.boolean(),
|
|
jump: z.boolean(),
|
|
data: z.boolean(),
|
|
read: z.boolean(),
|
|
write: z.boolean(),
|
|
indirect: z.boolean(),
|
|
computed: z.boolean(),
|
|
conditional: z.boolean(),
|
|
terminal: z.boolean(),
|
|
primary: z.boolean(),
|
|
operand_index: z.number().int(),
|
|
external: z.boolean(),
|
|
})
|
|
.strict();
|
|
const referenceEdge = z
|
|
.object({
|
|
source_address: ghidraCanonicalAddressSchema,
|
|
target_address: ghidraCanonicalAddressSchema,
|
|
source_procedure: procedureIdentity.nullable(),
|
|
target_procedure: procedureIdentity.nullable(),
|
|
kind: referenceKind,
|
|
})
|
|
.strict();
|
|
const procedureReferences = z
|
|
.object({
|
|
procedure: procedureIdentity,
|
|
direction: z.enum(["incoming", "outgoing"]),
|
|
references: z.array(referenceEdge),
|
|
reference_kinds_available: z.boolean().default(true),
|
|
unresolved_calls: z
|
|
.array(
|
|
z.strictObject({
|
|
address: ghidraCanonicalAddressSchema,
|
|
reason: z.string(),
|
|
}),
|
|
)
|
|
.default([]),
|
|
})
|
|
.strict();
|
|
const procedureInfo = z
|
|
.object({
|
|
name: z.string().min(1),
|
|
entrypoint: ghidraCanonicalAddressSchema,
|
|
basicblock_count: z.number().int().min(0),
|
|
length: z.number().int().min(0),
|
|
signature: z.string().nullable(),
|
|
locals: z.array(localVariable),
|
|
classification,
|
|
body: ghidraFunctionBody,
|
|
})
|
|
.strict()
|
|
.superRefine((value, context) => {
|
|
if (
|
|
!functionBodyEntryAgrees(value.body, value.entrypoint) ||
|
|
(value.body.available &&
|
|
(value.length !== value.body.total_bytes ||
|
|
(!value.classification.external && !value.body.contains_entry)))
|
|
)
|
|
context.addIssue({
|
|
code: "custom",
|
|
message: "Ghidra function body disagrees with entry point or length",
|
|
});
|
|
});
|
|
|
|
const ghidraNativeApiBoundary = nativeApiBoundarySchema.superRefine(
|
|
(value, context) => {
|
|
if (!value.available) return;
|
|
if (value.provenance !== "ghidra-high-function")
|
|
context.addIssue({
|
|
code: "custom",
|
|
message:
|
|
"Ghidra native API observations require HighFunction provenance",
|
|
});
|
|
const addresses = value.jump_tables.flatMap((table) => [
|
|
table.dispatch_address,
|
|
...table.data_sources.map(({ address }) => address),
|
|
...table.mappings.map(({ target_address }) => target_address),
|
|
...table.default_targets.map(({ target_address }) => target_address),
|
|
]);
|
|
if (
|
|
addresses.some(
|
|
(address) => !ghidraCanonicalAddressSchema.safeParse(address).success,
|
|
)
|
|
)
|
|
context.addIssue({
|
|
code: "custom",
|
|
message:
|
|
"Ghidra native API observation contains a non-canonical address",
|
|
});
|
|
},
|
|
);
|
|
|
|
const ghidraNativeValueFlow = nativeValueFlowSchema.superRefine(
|
|
(value, context) => {
|
|
if (!value.available) return;
|
|
const operations = value.operations;
|
|
const addresses = operations.flatMap((operation) => [
|
|
operation.address,
|
|
...operation.inputs.flatMap(({ location }) =>
|
|
location === null ? [] : [location],
|
|
),
|
|
...(operation.output?.location == null
|
|
? []
|
|
: [operation.output.location]),
|
|
]);
|
|
if (
|
|
addresses.some(
|
|
(address) => !ghidraCanonicalAddressSchema.safeParse(address).success,
|
|
)
|
|
)
|
|
context.addIssue({
|
|
code: "custom",
|
|
message: "Ghidra p-code flow contains a non-canonical address",
|
|
});
|
|
const operationIds = new Set(operations.map(({ id }) => id));
|
|
if (operationIds.size !== operations.length)
|
|
context.addIssue({
|
|
code: "custom",
|
|
message: "Ghidra p-code operation identifiers are not unique",
|
|
});
|
|
if (
|
|
value.def_use.some(
|
|
({ definition, use }) =>
|
|
!operationIds.has(definition) || !operationIds.has(use),
|
|
) ||
|
|
value.effects.some(({ operation }) => !operationIds.has(operation))
|
|
)
|
|
context.addIssue({
|
|
code: "custom",
|
|
message: "Ghidra p-code relationship references a missing operation",
|
|
});
|
|
if (
|
|
value.truncated !==
|
|
(value.omitted_operations_lower_bound > 0 ||
|
|
value.known_omitted_inputs > 0 ||
|
|
value.known_omitted_edges > 0)
|
|
)
|
|
context.addIssue({
|
|
code: "custom",
|
|
message: "Ghidra p-code truncation flag contradicts omitted counts",
|
|
});
|
|
},
|
|
);
|
|
|
|
const ghidraFunctionDossier = functionDossierSchema
|
|
.extend({
|
|
procedure: functionDossierSchema.shape.procedure.safeExtend({
|
|
classification,
|
|
body: ghidraFunctionBody,
|
|
}),
|
|
native_api: ghidraNativeApiBoundary,
|
|
native_value_flow: ghidraNativeValueFlow,
|
|
})
|
|
.strict()
|
|
.superRefine((value, context) => {
|
|
const addresses = [
|
|
value.procedure.address,
|
|
...value.comments.map(({ address }) => address),
|
|
...value.callers.map(({ address }) => address),
|
|
...value.callees.map(({ address }) => address),
|
|
...value.incoming_references.flatMap(referenceAddresses),
|
|
...value.outgoing_references.flatMap(referenceAddresses),
|
|
...value.referenced_strings.flatMap(({ address, source_address }) => [
|
|
address,
|
|
source_address,
|
|
]),
|
|
...value.referenced_names.flatMap(({ address, source_address }) => [
|
|
address,
|
|
source_address,
|
|
]),
|
|
...value.basic_blocks.flatMap(({ start, end, successors }) => [
|
|
start,
|
|
end,
|
|
...successors,
|
|
]),
|
|
];
|
|
if (
|
|
addresses.some(
|
|
(address) => !ghidraCanonicalAddressSchema.safeParse(address).success,
|
|
)
|
|
)
|
|
context.addIssue({
|
|
code: "custom",
|
|
message: "Ghidra dossier contains a non-canonical address",
|
|
});
|
|
const identities = [
|
|
value.procedure,
|
|
...value.callers,
|
|
...value.callees,
|
|
...value.incoming_references.flatMap(referenceProcedures),
|
|
...value.outgoing_references.flatMap(referenceProcedures),
|
|
];
|
|
if (
|
|
identities.some(
|
|
(identity) =>
|
|
!classification.safeParse(identity.classification).success ||
|
|
!ghidraFunctionBody.safeParse(identity.body).success ||
|
|
!functionBodyEntryAgrees(identity.body, identity.address) ||
|
|
(identity.body.available &&
|
|
identity.classification?.external === false &&
|
|
!identity.body.contains_entry),
|
|
)
|
|
)
|
|
context.addIssue({
|
|
code: "custom",
|
|
message: "Ghidra dossier omitted function classification",
|
|
});
|
|
if (
|
|
value.procedure.locals.some(
|
|
(variable) => !localVariable.safeParse(variable).success,
|
|
) ||
|
|
[...value.incoming_references, ...value.outgoing_references].some(
|
|
({ kind }) => !referenceKind.safeParse(kind).success,
|
|
)
|
|
)
|
|
context.addIssue({
|
|
code: "custom",
|
|
message: "Ghidra dossier contains invalid provenance",
|
|
});
|
|
if (
|
|
!value.limitations.some((item) => /indirect|computed/u.test(item)) ||
|
|
!value.limitations.some((item) => /provider|Ghidra|Hopper/u.test(item))
|
|
)
|
|
context.addIssue({
|
|
code: "custom",
|
|
message: "Ghidra dossier omitted required uncertainty limitations",
|
|
});
|
|
});
|
|
const ghidraFunctionInstructionWindow = functionInstructionWindowSchema
|
|
.extend({ procedure: procedureIdentity })
|
|
.strict();
|
|
|
|
const resultSchemas = {
|
|
analyze_function: ghidraFunctionDossier,
|
|
procedure_assembly: z.string(),
|
|
procedure_callees: z.array(ghidraCanonicalAddressSchema),
|
|
procedure_callers: z.array(ghidraCanonicalAddressSchema),
|
|
procedure_info: procedureInfo,
|
|
procedure_pseudo_code: z.string().nullable(),
|
|
read_function_instructions: ghidraFunctionInstructionWindow,
|
|
inspect_native_instruction: nativeInstructionSchema,
|
|
inspect_native_data_type: nativeDataTypeSchema,
|
|
resolve_native_call_targets: nativeCallTargetsSchema,
|
|
procedure_references: procedureReferences,
|
|
xrefs: z.array(ghidraCanonicalAddressSchema),
|
|
} satisfies Readonly<Record<GhidraFunctionOperation, z.ZodType>>;
|
|
|
|
/** Require exact, bounded Java-bridge function output before creating Evidence. */
|
|
export const parseGhidraFunctionResult = (
|
|
operation: GhidraFunctionOperation,
|
|
value: JsonValue,
|
|
): Result<JsonValue, AnalysisOutputError> => {
|
|
const parsed = resultSchemas[operation].safeParse(value);
|
|
return parsed.success
|
|
? ok(parsed.data)
|
|
: err(
|
|
new AnalysisOutputError(
|
|
operation,
|
|
"Ghidra bridge output did not match the function-analysis contract",
|
|
{ cause: parsed.error },
|
|
),
|
|
);
|
|
};
|
|
|
|
const referenceAddresses = (
|
|
edge: FunctionDossier["incoming_references"][number],
|
|
): readonly string[] => [
|
|
edge.source_address,
|
|
edge.target_address,
|
|
...(edge.source_procedure === null ? [] : [edge.source_procedure.address]),
|
|
...(edge.target_procedure === null ? [] : [edge.target_procedure.address]),
|
|
];
|
|
|
|
const referenceProcedures = (
|
|
edge: FunctionDossier["incoming_references"][number],
|
|
): readonly NonNullable<
|
|
FunctionDossier["incoming_references"][number]["source_procedure"]
|
|
>[] => [
|
|
...(edge.source_procedure === null ? [] : [edge.source_procedure]),
|
|
...(edge.target_procedure === null ? [] : [edge.target_procedure]),
|
|
];
|