OCI Policy Analysis Application Core

The core reference covers the maintained models, parsers, repositories, services, and engines used by the CLI, UI, web application, and MCP server.

Models

Legacy re-exports for model modules (compatibility shim).

IAM entity and search models (groups, users, dynamic groups, compartments).

class oci_policy_analysis.application.core.models.models_iam.Compartment[source]

Bases: TypedDict

Model representing an OCI compartment (identity and metadata). Captures id, name, parent, description, path, lifecycle, and tags where available. Adds optional analysis/derived fields for policy statement counts.

Optional/derived fields (set after loading and analysis):

statement_count_direct: Number of policy statements directly in this compartment.
statement_count_cumulative: Cumulative policy statements (this + all ancestors in path).

id: Annotated[str, 'Compartment OCID']

name: Annotated[str, 'Display name of the compartment']

parent_id: Annotated[str, 'Parent compartment OCID']

hierarchy_path: Annotated[str, 'Full compartment hierarchy path, e.g., "ROOT/HR/Payroll"']

description: NotRequired[Annotated[str, 'Description of the compartment']]

lifecycle_state: NotRequired[Annotated[str, 'Lifecycle state (e.g., ACTIVE, DELETED)']]

tags: NotRequired[Annotated[dict[str, str], 'Optional. All freeform and defined tags associated with the compartment.']]

statement_count_direct: NotRequired[Annotated[int, 'Number of policy statements directly in this compartment (analysis-derived, optional)']]

statement_count_cumulative: NotRequired[Annotated[int, 'Cumulative number of policy statements including all ancestors (analysis-derived, optional)']]

class oci_policy_analysis.application.core.models.models_iam.Group[source]

Bases: TypedDict

OCI IAM group.

domain_name: NotRequired[Annotated[str, 'Identity domain; defaults to Default.']]

group_name: Annotated[str, 'Group name.']

group_id: NotRequired[Annotated[str, 'Group ID.']]

group_ocid: NotRequired[Annotated[str, 'Group OCID.']]

description: NotRequired[Annotated[str, 'Group description.']]

class oci_policy_analysis.application.core.models.models_iam.User[source]

Bases: TypedDict

OCI IAM user.

domain_name: NotRequired[Annotated[str, 'Identity domain; defaults to Default.']]

user_name: Annotated[str, 'User name.']

user_ocid: NotRequired[Annotated[str, 'User OCID.']]

display_name: NotRequired[Annotated[str, 'Display name.']]

email: NotRequired[Annotated[str, 'Primary email.']]

user_id: NotRequired[Annotated[str, 'User ID.']]

groups: NotRequired[Annotated[list[str], 'Group OCIDs.']]

class oci_policy_analysis.application.core.models.models_iam.ConditionAtom[source]

Bases: TypedDict

Parsed condition atom for display/evidence.

id: Annotated[str, 'Atom id within the clause.']

left: Annotated[str, 'Left side expression.']

operator: Annotated[str, 'Operator.']

right: Annotated[str, 'Right side expression.']

normalized_left: Annotated[str, 'Case-normalized left side.']

value_type: Annotated[str, 'Right side value kind.']

evidence_kind: Annotated[str, 'Evidence category.']

subexpression: Annotated[str, 'Raw atom text.']

class oci_policy_analysis.application.core.models.models_iam.ConditionStructure[source]

Bases: TypedDict

Parsed rule/where structure for display.

raw_text: Annotated[str, 'Raw clause text.']

parse_status: Annotated[Literal['parsed', 'partial', 'unsupported', 'absent'], 'Parse status.']

structure: Annotated[str, 'Compact tree summary.']

atoms: Annotated[list[ConditionAtom], 'Flat condition atoms.']

residual_text: Annotated[str, 'Unparsed or residual text.']

warnings: Annotated[list[str], 'Parse warnings.']

class oci_policy_analysis.application.core.models.models_iam.DynamicGroup[source]

Bases: TypedDict

OCI dynamic group.

domain_name: NotRequired[Annotated[str, 'Identity domain; defaults to Default.']]

domain_ocid: NotRequired[Annotated[str, 'Domain OCID.']]

dynamic_group_name: Annotated[str, 'Dynamic group name.']

dynamic_group_ocid: NotRequired[Annotated[str, 'Dynamic group OCID.']]

dynamic_group_id: NotRequired[Annotated[str, 'Dynamic group ID.']]

matching_rule: NotRequired[Annotated[str, 'Dynamic group matching rule.']]

matching_rule_parsed_structure: NotRequired[Annotated[str, 'Parsed matching-rule summary.']]

matching_rule_elements: NotRequired[Annotated[str, 'Parsed matching-rule atoms for display.']]

matching_rule_structure: NotRequired[Annotated[ConditionStructure, 'Parsed matching-rule structure.']]

description: NotRequired[Annotated[str | None, 'Dynamic group description.']]

in_use: NotRequired[Annotated[bool, 'True when referenced by policy.']]

creation_time: NotRequired[Annotated[str, 'Creation time.']]

created_by_ocid: NotRequired[Annotated[str, 'Creator OCID.']]

created_by_name: NotRequired[Annotated[str, 'Creator name.']]

class oci_policy_analysis.application.core.models.models_iam.GroupSearch[source]

Bases: TypedDict

Group search filters.

domain_name: Annotated[list[str], 'Domain names to match.']

group_name: Annotated[list[str], 'Group names or substrings.']

group_ocid: Annotated[list[str], 'Group OCIDs or substrings.']

class oci_policy_analysis.application.core.models.models_iam.UserSearch[source]

Bases: TypedDict

User search filters.

domain_name: Annotated[list[str], 'Domain names to match.']

search: Annotated[list[str], 'User names/display names or substrings.']

user_ocid: Annotated[str, 'User OCIDs or substrings.']

class oci_policy_analysis.application.core.models.models_iam.DynamicGroupSearch[source]

Bases: TypedDict

Dynamic group search filters.

domain_name: NotRequired[Annotated[list[str], 'Domain names to match.']]

dynamic_group_name: NotRequired[Annotated[list[str], 'Dynamic group names or substrings.']]

matching_rule: NotRequired[Annotated[list[str], 'Matching rule substrings.']]

dynamic_group_ocid: Annotated[str, 'Dynamic group OCIDs or substrings.']

in_use: NotRequired[Annotated[bool, 'Filter by policy reference status.']]

Policy and statement models.

class oci_policy_analysis.application.core.models.models_policy.Principal[source]

Bases: TypedDict

Canonical policy principal.

principal_type: Annotated[str, 'Principal type.']

principal_key: Annotated[str, 'Canonical principal key.']

domain_name: NotRequired[Annotated[str | None, 'Identity domain.']]

name: NotRequired[Annotated[str, 'Principal name.']]

ocid: NotRequired[Annotated[str, 'Principal OCID.']]

display_name: NotRequired[Annotated[str, 'Display label.']]

resource_type: NotRequired[Annotated[str, 'Resource principal type from request.principal.type.']]

resource_ocid: NotRequired[Annotated[str, 'Resource principal OCID from request.principal.id.']]

compartment_ocid: NotRequired[Annotated[str, 'Resource principal compartment OCID.']]

resource_compartment_ocid: NotRequired[Annotated[str, 'Resource principal compartment OCID.']]

workload_namespace: NotRequired[Annotated[str, 'OKE workload identity namespace.']]

workload_service_account: NotRequired[Annotated[str, 'OKE workload identity service account.']]

workload_cluster_id: NotRequired[Annotated[str, 'OKE workload identity cluster OCID.']]

match_mode: NotRequired[Annotated[str, 'Principal match mode hint.']]

class oci_policy_analysis.application.core.models.models_policy.ConditionAtom[source]

Bases: TypedDict

Parsed condition atom for display/evidence.

id: Annotated[str, 'Atom id within the clause.']

left: Annotated[str, 'Left side expression.']

operator: Annotated[str, 'Operator.']

right: Annotated[str, 'Right side expression.']

normalized_left: Annotated[str, 'Case-normalized left side.']

value_type: Annotated[str, 'Right side value kind.']

evidence_kind: Annotated[str, 'Evidence category.']

subexpression: Annotated[str, 'Raw atom text.']

class oci_policy_analysis.application.core.models.models_policy.ConditionStructure[source]

Bases: TypedDict

Parsed where/rule structure for display.

raw_text: Annotated[str, 'Raw clause text.']

parse_status: Annotated[Literal['parsed', 'partial', 'unsupported', 'absent'], 'Parse status.']

structure: Annotated[str, 'Compact tree summary.']

atoms: Annotated[list[ConditionAtom], 'Flat condition atoms.']

residual_text: Annotated[str, 'Unparsed or residual text.']

warnings: Annotated[list[str], 'Parse warnings.']

class oci_policy_analysis.application.core.models.models_policy.BasePolicy[source]

Bases: TypedDict

Model representing an OCI IAM policy (identity/metadata).

policy_name: Annotated[str, 'The name of the policy.']

policy_ocid: Annotated[str, 'The OCID of the policy. Not required for filters.']

description: Annotated[str | None, 'The description of the policy. Not required for filters.']

compartment_ocid: Annotated[str, 'The OCID of the compartment containing the policy. Not required for filters.']

compartment_path: Annotated[str, 'Full compartment path string for the policy (e.g., "ROOT/CompA/CompB").']

creation_time: Annotated[str, 'The creation time of the policy. Not required for filters.']

tags: NotRequired[Annotated[dict[str, str], 'Optional. Flattened tag map for display only (freeform plus defined flattened as "namespace:key").']]

freeform_tags: NotRequired[Annotated[dict[str, str], 'Optional. Freeform tag map as stored in OCI.']]

defined_tags: NotRequired[Annotated[dict[str, dict[str, str]], 'Optional. Defined tag map as stored in OCI (namespace -> key -> value).']]

class oci_policy_analysis.application.core.models.models_policy.PolicySearch[source]

Bases: TypedDict

Policy statement filters.

action: Annotated[list[str], "Statement actions: 'allow' or 'deny'."]

exact_groups: Annotated[list[Group], 'Exact groups.']

exact_users: Annotated[list[User], 'Exact users.']

exact_dynamic_groups: Annotated[list[DynamicGroup], 'Exact dynamic groups.']

search_groups: Annotated[GroupSearch, 'Fuzzy group search.']

search_users: Annotated[UserSearch, 'Fuzzy user search.']

search_dynamic_groups: Annotated[DynamicGroupSearch, 'Fuzzy dynamic group search.']

principals: Annotated[list[Principal], 'Structured principal selectors.']

principal: Annotated[Principal, 'Single structured principal selector.']

principal_keys: Annotated[list[str], 'Canonical principal keys.']

verb: Annotated[list[Literal['inspect', 'read', 'use', 'manage']], 'Policy verbs to match.']

statement_text: Annotated[list[str], 'Statement text substrings.']

policy_name: Annotated[list[str], 'Policy names.']

compartment_path: Annotated[list[str], "Policy compartment paths; 'ROOTONLY' means root policies."]

resource: Annotated[list[str], 'Policy resource types.']

location: Annotated[list[str], 'Policy locations or OCIDs.']

effective_path: Annotated[list[str], 'Effective paths; exact or prefix match.']

subject_type: Annotated[list[Literal['group', 'dynamic-group', 'any-user', 'any-group', 'service']], 'Policy subject types.']

subject: Annotated[list[str], 'Subject identifiers.']

principal_key: Annotated[list[str], 'Legacy principal key filter.']

permission: Annotated[list[str], 'Permission names.']

comments: Annotated[list[str], 'Statement comment substrings.']

conditions: Annotated[list[str], 'Condition substrings.']

tag_access_type: Annotated[list[str], 'Parsed tag condition access type filters.']

tag_access_semantics: Annotated[list[str], 'Parsed tag condition semantic access filters.']

tag_namespace: Annotated[list[str], 'Parsed tag namespace filters.']

tag_key: Annotated[list[str], 'Parsed tag key filters.']

tag_value: Annotated[list[str], 'Parsed tag value filters.']

tag_operator: Annotated[list[str], 'Parsed tag operator filters.']

condition_atom_terms: Annotated[list[str], 'Terms searched across parsed condition atom fields.']

policy_tag: Annotated[list[str], 'Terms searched across policy object tags.']

policy_defined_tag: Annotated[list[str], 'Terms searched across policy object defined tags.']

policy_freeform_tag: Annotated[list[str], 'Terms searched across policy object freeform tags.']

valid: Annotated[bool, 'True for valid statements; false for invalid.']

class oci_policy_analysis.application.core.models.models_policy.PolicyOverlap[source]

Bases: TypedDict

Model for representing overlap/conflict analysis between policy statements.

superseded_by: Annotated[str, 'The policy name that supersedes this statement']

confidence: Annotated[str, 'Confidence level of the overlap (e.g., "high", "medium", "low")']

reason: Annotated[str, 'Explanation for the overlap detection']

statement_text: Annotated[str, 'The statement text of the superseding statement']

internal_id: Annotated[str, 'The internal ID of the superseding statement']

permission_overlap: Annotated[list[str], 'List of specific permissions that overlap between the two statements']

additional_notes: NotRequired[Annotated[str, 'Any additional notes about the overlap analysis.']]

class oci_policy_analysis.application.core.models.models_policy.BasePolicyStatement[source]

Bases: TypedDict

Base model for all OCI policy statement types, containing shared fields.

policy_name: Annotated[str, 'Policy name.']

policy_ocid: Annotated[str, 'Policy OCID.']

compartment_ocid: Annotated[str, 'Policy compartment OCID.']

compartment_path: Annotated[str, 'Policy compartment path.']

statement_text: Annotated[str, 'Raw policy statement.']

creation_time: Annotated[str, 'Policy creation time.']

internal_id: Annotated[str, 'Internal statement ID.']

parsed: Annotated[bool, 'True when parsed.']

class oci_policy_analysis.application.core.models.models_policy.DefineStatement[source]

Bases: BasePolicyStatement

Parsed OCI IAM ‘define’ policy statement with optional metadata.

valid: Annotated[bool, 'True when valid.']

defined_type: Annotated[str, 'Defined object type.']

defined_name: Annotated[str, 'Defined object name.']

ocid_alias: Annotated[str, 'Defined OCID alias.']

comment: NotRequired[Annotated[str, 'Statement comment.']]

policy_name: Annotated[str, 'Policy name.']

policy_ocid: Annotated[str, 'Policy OCID.']

compartment_ocid: Annotated[str, 'Policy compartment OCID.']

compartment_path: Annotated[str, 'Policy compartment path.']

statement_text: Annotated[str, 'Raw policy statement.']

creation_time: Annotated[str, 'Policy creation time.']

internal_id: Annotated[str, 'Internal statement ID.']

parsed: Annotated[bool, 'True when parsed.']

class oci_policy_analysis.application.core.models.models_policy.EndorseStatement[source]

Bases: BasePolicyStatement

Parsed OCI IAM ‘endorse’ cross-tenancy policy statement with optional metadata.

valid: Annotated[bool, 'True when valid.']

action_type: Annotated[Literal['endorse', 'deny endorse'], 'Type of endorse action: either "endorse" or "deny endorse"']

endorsed_principal_type: Annotated[Literal['group', 'dynamic-group', 'any-user', 'any-group', 'service'], 'Type of principal: group, dynamic-group, any-user, service.']

principal_keys: NotRequired[Annotated[list[str], 'Canonical principal keys.']]

endorsed_principal: Annotated[str, 'Endorsed principal.']

endorsed_principal_tenancy: Annotated[str, 'Endorsed principal tenancy.']

endorse_action: Annotated[str, 'Endorse verb/permission.']

endorse_resource: Annotated[str, 'Endorse resource.']

endorse_permissions: NotRequired[Annotated[list[str], 'Endorse permissions.']]

endorse_tenancy: Annotated[str, 'Endorsed tenancy alias.']

endorse_associate_resource: NotRequired[Annotated[str, 'Associate resource A.']]

endorse_associate_tenancy: NotRequired[Annotated[str, 'Associate tenancy A.']]

endorse_associate_with_resource: NotRequired[Annotated[str, 'Associate resource B.']]

endorse_associate_with_tenancy: NotRequired[Annotated[str, 'Associate tenancy B.']]

associate_clause_raw: NotRequired[Annotated[str, 'Raw associate clause.']]

tenancy_aliases: NotRequired[Annotated[list[str], 'Referenced tenancy aliases.']]

resolved_aliases: NotRequired[Annotated[dict[str, dict[str, str | bool]], 'Alias resolution map populated by intelligence step; keys are alias names and values include ocid/resolved.']]

aliases_resolved: NotRequired[Annotated[bool, 'True when aliases resolve.']]

where_clause: NotRequired[Annotated[str, 'Where clause.']]

comment: NotRequired[Annotated[str, 'Statement comment.']]

policy_name: Annotated[str, 'Policy name.']

policy_ocid: Annotated[str, 'Policy OCID.']

compartment_ocid: Annotated[str, 'Policy compartment OCID.']

compartment_path: Annotated[str, 'Policy compartment path.']

statement_text: Annotated[str, 'Raw policy statement.']

creation_time: Annotated[str, 'Policy creation time.']

internal_id: Annotated[str, 'Internal statement ID.']

parsed: Annotated[bool, 'True when parsed.']

class oci_policy_analysis.application.core.models.models_policy.AdmitStatement[source]

Bases: BasePolicyStatement

Parsed OCI IAM ‘admit’ cross-tenancy policy statement with parsed metadata.

valid: Annotated[bool, 'True when valid.']

action_type: Annotated[Literal['admit', 'deny admit'], 'Type of admit action: either "admit" or "deny admit"']

admitted_principal_type: Annotated[Literal['group', 'dynamic-group', 'any-user', 'any-group', 'service'], 'Type of principal: group, dynamic-group, any-user, etc.']

principal_keys: NotRequired[Annotated[list[str], 'Canonical principal keys.']]

admitted_principal: Annotated[str, 'Admitted principal.']

admitted_tenancy: Annotated[str, 'Admitted tenancy.']

admit_action: Annotated[str, 'Admit verb/permission.']

admit_resource: Annotated[str, 'Admit resource.']

admit_permissions: NotRequired[Annotated[list[str], 'Admit permissions.']]

admit_location_type: Annotated[str, 'Admit location type.']

admit_location: Annotated[str, 'Admit location.']

admit_associate_resource: NotRequired[Annotated[str, 'Associate resource A.']]

admit_associate_tenancy: NotRequired[Annotated[str, 'Associate tenancy A.']]

admit_associate_with_resource: NotRequired[Annotated[str, 'Associate resource B.']]

admit_associate_with_tenancy: NotRequired[Annotated[str, 'Associate tenancy B.']]

associate_clause_raw: NotRequired[Annotated[str, 'Raw associate clause.']]

tenancy_aliases: NotRequired[Annotated[list[str], 'Referenced tenancy aliases.']]

resolved_aliases: NotRequired[Annotated[dict[str, dict[str, str | bool]], 'Alias resolution map populated by intelligence step; keys are alias names and values include ocid/resolved.']]

aliases_resolved: NotRequired[Annotated[bool, 'True when aliases resolve.']]

where_clause: NotRequired[Annotated[str, 'Where clause.']]

comment: NotRequired[Annotated[str, 'Statement comment.']]

policy_name: Annotated[str, 'Policy name.']

policy_ocid: Annotated[str, 'Policy OCID.']

compartment_ocid: Annotated[str, 'Policy compartment OCID.']

compartment_path: Annotated[str, 'Policy compartment path.']

statement_text: Annotated[str, 'Raw policy statement.']

creation_time: Annotated[str, 'Policy creation time.']

internal_id: Annotated[str, 'Internal statement ID.']

parsed: Annotated[bool, 'True when parsed.']

class oci_policy_analysis.application.core.models.models_policy.RegularPolicyStatement[source]

Bases: BasePolicyStatement

Represents a parsed OCI IAM policy statement.

action: Annotated[Literal['allow', 'deny'], "Statement action: 'allow' or 'deny'."]

valid: Annotated[bool, 'True when valid.']

invalid_reasons: Annotated[list[str], 'Validation errors.']

subject_type: Annotated[str, 'Policy subject type.']

subject: Annotated[list[tuple[str | None, str]] | str, 'Policy subject value.']

principals: Annotated[list[Principal], 'Canonical principals.']

principal_keys: NotRequired[Annotated[list[str], 'Canonical principal keys.']]

verb: Annotated[str, 'IAM verb.']

resource: Annotated[str, 'Policy resource type.']

permission: Annotated[list[str], 'Derived permission names.']

location_type: Annotated[str, 'Location resolution type.']

location: Annotated[str, 'Policy location.']

effective_compartment_ocid: Annotated[str | None, 'Effective compartment OCID.']

effective_path: Annotated[str | None, 'Effective compartment path.']

conditions: Annotated[str, 'Statement conditions.']

conditions_where_clause: NotRequired[Annotated[str, 'Raw conditions/where-clause text.']]

conditions_parsed_structure: NotRequired[Annotated[str, 'Parsed condition tree summary.']]

conditions_elements: NotRequired[Annotated[str, 'Parsed condition atoms for display.']]

condition_atoms: NotRequired[Annotated[list[ConditionAtom], 'Parsed condition atoms.']]

principal_evidence: NotRequired[Annotated[list[ConditionAtom], 'Condition atoms used as principal identity evidence.']]

residual_conditions: NotRequired[Annotated[list[ConditionAtom], 'Condition atoms not used as principal identity evidence.']]

match_confidence: NotRequired[Annotated[str, 'Workload principal match confidence.']]

match_confidence_reason: NotRequired[Annotated[str, 'Explanation for workload principal match confidence.']]

confidence: NotRequired[Annotated[str, 'Display confidence alias.']]

where_clause: NotRequired[Annotated[ConditionStructure, 'Parsed where-clause structure.']]

where_clause_structure: NotRequired[Annotated[ConditionStructure, 'Parsed where-clause structure.']]

tag_conditions: NotRequired[Annotated[list[dict[str, object]], 'Parsed tag condition details.']]

tag_context_warnings: NotRequired[Annotated[list[str], 'Contextual tag-based access warnings.']]

comments: Annotated[str, 'Statement comments.']

parsing_notes: Annotated[list[str], 'Parser notes.']

policy_name: Annotated[str, 'Policy name.']

policy_ocid: Annotated[str, 'Policy OCID.']

compartment_ocid: Annotated[str, 'Policy compartment OCID.']

compartment_path: Annotated[str, 'Policy compartment path.']

statement_text: Annotated[str, 'Raw policy statement.']

creation_time: Annotated[str, 'Policy creation time.']

internal_id: Annotated[str, 'Internal statement ID.']

parsed: Annotated[bool, 'True when parsed.']

class oci_policy_analysis.application.core.models.models_policy.PolicySummary[source]

Bases: TypedDict

Policy filter summary.

response_type: Literal['summary']

total_statements: Annotated[int, 'Matched statement count.']

truncated: Annotated[bool, 'True when summarized.']

truncation_point: Annotated[int, 'Full-result threshold.']

policy_breakdown: Annotated[dict[str, int], 'Count by policy.']

action_breakdown: Annotated[dict[str, int], 'Count by action.']

compartment_breakdown: Annotated[dict[str, int], 'Count by compartment.']

subject_type_breakdown: Annotated[dict[str, int], 'Count by subject type.']

verb_breakdown: Annotated[dict[str, int], 'Count by verb.']

sample_statements: Annotated[list[str], 'Sample statements.']

message: Annotated[str, 'Summary note.']

class oci_policy_analysis.application.core.models.models_policy.PolicyStatementFull[source]

Bases: TypedDict

Policy filter full result.

response_type: Literal['full']

statements: Annotated[list[RegularPolicyStatement], 'Matched statements.']

total_count: Annotated[int, 'Returned statement count.']

class oci_policy_analysis.application.core.models.models_policy.PolicyIntelligence[source]

Bases: TypedDict

Model for high-level analytics and findings of policy analysis (IAM intelligence overlay).

overlaps: list[dict]

recommendations: list[dict]

risk_scores: list[dict]

consolidations: list[dict]

cleanup_items: NotRequired[dict]

class oci_policy_analysis.application.core.models.models_consolidation.ProtectedStatementReference[source]

Bases: TypedDict

Unique reference for a protected statement, captures context for later reconciliation, supports fuzzy/orphan recovery.

internal_id: Annotated[str, 'Hash ID of policy statement (may change if user edits statement)']

policy_ocid: Annotated[str, 'OCI Policy OCID this statement belongs to']

policy_name: Annotated[str, 'Human policy name']

statement_text: Annotated[str, 'Canonical policy statement text at time of protection']

added_at: NotRequired[Annotated[str, 'Timestamp when statement was protected (ISO-8601)']]

class oci_policy_analysis.application.core.models.models_consolidation.ProtectedStatementSet[source]

Bases: TypedDict

Canonical set of protected (non-consolidatable) statements for a session/project. Tied to a tenancy for context, supports load validation and orphan reporting.

tenancy_ocid: Annotated[str, 'Tenancy OCID for this policy set']

dataset_version: NotRequired[Annotated[str, 'Version label or hash for this snapshot of policy set']]

protected: Annotated[list[ProtectedStatementReference], 'List of protected statements (with context)']

orphaned_internal_ids: NotRequired[Annotated[list[str], 'Protected internal_ids not found on load']]

last_updated: NotRequired[Annotated[str, 'Timestamp']]

class oci_policy_analysis.application.core.models.models_consolidation.CandidateStatementReference[source]

Bases: TypedDict

Reference for a candidate statement nominated for consolidation.

internal_id: Annotated[str, 'Hash/internal ID']

policy_ocid: Annotated[str, 'Policy OCID']

statement_text: Annotated[str, 'Policy statement text at time of nomination']

nominated_at: NotRequired[Annotated[str, 'Timestamp']]

class oci_policy_analysis.application.core.models.models_consolidation.CandidateSelectionSet[source]

Bases: TypedDict

Explicit set of statements selected as candidates for consolidation (per session/project).

tenancy_ocid: Annotated[str, 'Tenancy OCID']

candidates: Annotated[list[CandidateStatementReference], 'Current set of candidate statements']

orphaned_internal_ids: NotRequired[Annotated[list[str], 'Candidates not found on load']]

last_updated: NotRequired[Annotated[str, 'Timestamp']]

class oci_policy_analysis.application.core.models.models_consolidation.SkippedStatement[source]

Bases: TypedDict

A candidate statement that the strategy did not include in the plan (e.g. does not fit strategy rules).

internal_id: Annotated[str, 'Statement internal_id that was skipped']

reason: Annotated[str, 'Why this statement was not applied (e.g. "Does not match strategy scope")']

statement_text: NotRequired[Annotated[str, 'Snippet or full text for display']]

class oci_policy_analysis.application.core.models.models_consolidation.PlanStep[source]

Bases: TypedDict

Single step in a consolidation plan: add, modify, or delete a policy, or change tags/statements. Each step may track before/after, tags, rollback, and execution result.

step_id: Annotated[str, 'Unique plan step identifier']

action: Annotated[Literal['add', 'modify', 'delete'], 'Plan step action type']

policy_ocid: Annotated[str, 'Target policy OCID']

before_statements: Annotated[list[str], 'Statements before this step is applied']

after_statements: Annotated[list[str], 'Statements after this step is applied']

before_tags: Annotated[dict[str, str], 'Tag state before step']

after_tags: Annotated[dict[str, str], 'Desired tag state after step']

plan_tags: NotRequired[Annotated[dict[str, str], 'Plan-specific tags to mark plan/step']]

executed: NotRequired[Annotated[bool, 'Whether this step has been executed/applied']]

executed_at: NotRequired[Annotated[str, 'Timestamp of execution']]

execution_status: NotRequired[Annotated[Literal['PENDING', 'COMPLETE', 'ROLLED_BACK', 'DRIFTED'], 'Current status']]

execution_notes: NotRequired[Annotated[str, 'API/OCI result notes, error etc.']]

rollback_command: NotRequired[Annotated[str, 'Command or instruction to roll back this step']]

location_change_notes: NotRequired[Annotated[list[str], 'Notes when statement location was rewritten due to policy move (e.g. compartment X to A:B).']]

compartment_ocid: NotRequired[Annotated[str, 'Compartment OCID (add: where to create; delete: original)']]

create_policy_name: NotRequired[Annotated[str, 'Policy name (add: new; delete: original, for display/rollback)']]

create_policy_description: NotRequired[Annotated[str, 'Description for the new policy (add step only); user may change.']]

class oci_policy_analysis.application.core.models.models_consolidation.ConsolidationPlan[source]

Bases: TypedDict

Full consolidation plan: tenancy, steps, plan label, execution log. Supports before/after tags/statements, per-policy, per-step.

plan_id: Annotated[str, 'Unique ID for this plan']

tenancy_ocid: Annotated[str, 'Tenancy OCID']

plan_label: Annotated[str, 'User label for this plan (optional)']

created_at: Annotated[str, 'Creation timestamp']

plan_steps: Annotated[list[PlanStep], 'Full set of steps for the plan']

execution_log: NotRequired[Annotated[list[str], 'History of execution attempts']]

plan_tags: NotRequired[Annotated[dict[str, str], 'Tags used to mark plan/steps in OCI']]

notes: NotRequired[Annotated[str, 'Free-form plan notes (strategy or user)']]

skipped_statements: NotRequired[Annotated[list[SkippedStatement], 'Candidates not applied by this strategy, with reason']]

class oci_policy_analysis.application.core.models.models_consolidation.PlanExecutionCheckResult[source]

Bases: TypedDict

Outcome of reloading policy data and checking/tagging plan execution.

step_id: Annotated[str, 'The step being checked']

policy_ocid: Annotated[str, 'Policy being checked']

executed: Annotated[bool, 'Does current policy/tag state match after_*?']

missing: NotRequired[Annotated[bool, 'Policy or tag missing on reload']]

drifted: NotRequired[Annotated[bool, 'Present but data does not match expected']]

execution_notes: NotRequired[Annotated[str, 'Details of match/mismatch']]

checked_at: Annotated[str, 'When check was performed']

class oci_policy_analysis.application.core.models.models_consolidation.ConsolidationSessionAuditEntry[source]

Bases: TypedDict

Audit log for session/project, tracking user and system actions for transparency, rollback, and reporting.

timestamp: Annotated[str, 'Timestamp of action']

user: Annotated[str, 'User or system role']

action: Annotated[str, 'What was performed (UI selection, plan create, execute step, rollback, etc)']

details: NotRequired[Annotated[dict, 'Additional context']]

class oci_policy_analysis.application.core.models.models_consolidation.ConsolidationSession[source]

Bases: TypedDict

Links together a persistent session: tenancy, protection, candidates, plan, and audit. This is the root object to (de)serialize to disk/cloud for long-lived projects.

tenancy_ocid: Annotated[str, 'Tenancy OCID']

dataset_version: NotRequired[Annotated[str, 'Loaded data version']]

protected_set: Annotated[ProtectedStatementSet, 'Set of protected statements']

candidate_set: Annotated[CandidateSelectionSet, 'Selected statements to consolidate']

plan: Annotated[ConsolidationPlan, 'Proposed or active plan']

audit_log: NotRequired[Annotated[list[ConsolidationSessionAuditEntry], 'Full session audit log']]

execution_results: NotRequired[Annotated[dict[str, PlanExecutionCheckResult], 'Most recent per-step execution check']]

Search response models for IAM entities and reference data.

class oci_policy_analysis.application.core.models.models_responses.UserSummary[source]

Bases: TypedDict

User search summary.

response_type: Literal['summary']

total_users: Annotated[int, 'Matched user count.']

truncated: Annotated[bool, 'True when summarized.']

truncation_point: Annotated[int, 'Full-result threshold.']

domain_breakdown: Annotated[dict[str, int], 'User count by domain.']

sample_users: Annotated[list[str], 'Sample user names.']

message: Annotated[str, 'Summary note.']

class oci_policy_analysis.application.core.models.models_responses.UserSearchFull[source]

Bases: TypedDict

Full user search result.

response_type: Literal['full']

users: Annotated[list[User], 'Matched users.']

total_count: Annotated[int, 'Returned user count.']

class oci_policy_analysis.application.core.models.models_responses.GroupSummary[source]

Bases: TypedDict

Group search summary.

response_type: Literal['summary']

total_groups: Annotated[int, 'Matched group count.']

truncated: Annotated[bool, 'True when summarized.']

truncation_point: Annotated[int, 'Full-result threshold.']

domain_breakdown: Annotated[dict[str, int], 'Group count by domain.']

sample_groups: Annotated[list[str], 'Sample group names.']

message: Annotated[str, 'Summary note.']

class oci_policy_analysis.application.core.models.models_responses.GroupSearchFull[source]

Bases: TypedDict

Full group search result.

response_type: Literal['full']

groups: Annotated[list[Group], 'Matched groups.']

total_count: Annotated[int, 'Returned group count.']

class oci_policy_analysis.application.core.models.models_responses.DynamicGroupSummary[source]

Bases: TypedDict

Dynamic group search summary.

response_type: Literal['summary']

total_dynamic_groups: Annotated[int, 'Matched dynamic group count.']

truncated: Annotated[bool, 'True when summarized.']

truncation_point: Annotated[int, 'Full-result threshold.']

domain_breakdown: Annotated[dict[str, int], 'Dynamic group count by domain.']

in_use_breakdown: Annotated[dict[str, int], 'Count by in-use status.']

sample_dynamic_groups: Annotated[list[str], 'Sample dynamic group names.']

message: Annotated[str, 'Summary note.']

class oci_policy_analysis.application.core.models.models_responses.DynamicGroupSearchFull[source]

Bases: TypedDict

Full dynamic group search result.

response_type: Literal['full']

dynamic_groups: Annotated[list[DynamicGroup], 'Matched dynamic groups.']

total_count: Annotated[int, 'Returned dynamic group count.']

class oci_policy_analysis.application.core.models.models_responses.ReferenceDataDiffResult[source]

Bases: TypedDict

Result model describing the outcome of comparing two cached reference data sets.

response_type: Literal['reference_data_diff']

cache_a: Annotated[str, 'Name of older cache (file or key)']

cache_b: Annotated[str, 'Name of newer cache (file or key)']

diff_summary: Annotated[str, 'One-line or short summary of differences (added, changed, removed)']

diff_details: Annotated[dict, 'DeepDiff result details or filtered view suitable for UI display']

message: Annotated[str, 'Human-readable message about the diff result or info']

class oci_policy_analysis.application.core.models.models_simulation.SimulationPrepareRequest[source]

Bases: TypedDict

Canonical input for simulation preparation.

Used to obtain principal and where-clause context, never for actual simulation execution.

compartment_path: Annotated[str, 'Effective compartment path, e.g. "ROOT/Finance"']

principal_type: Annotated[Literal['group', 'user', 'dynamic-group', 'any-user', 'any-group', 'service'], 'Principal type for simulation']

principal: Annotated[str | list[str | None], 'String for "any-user"/"any-group"/"service"; [domain, name] for user/group/dynamic-group']

class oci_policy_analysis.application.core.models.models_simulation.SimulationPrepareResponse[source]

Bases: TypedDict

Output of simulation preparation.

Provides required where fields and a standardized principal key.

required_where_fields: Annotated[list[str], 'All unique where clause variable names required for input']

principal_key: Annotated[str, 'Principal key as calculated by engine (e.g., "user:Default/anita")']

class oci_policy_analysis.application.core.models.models_simulation.SimulationScenario[source]

Bases: TypedDict

Canonical input for a single simulation scenario.

Batch-mode simulations use a list of these in SimulationBatchRequest.

compartment_path: Annotated[str, 'Effective compartment path']

scenario_internal_id: NotRequired[Annotated[str, 'Optional scenario identifier for grouping related simulations']]

scenario_name: NotRequired[Annotated[str, 'Optional human-readable scenario name']]

principal_key: Annotated[str, 'Principal key, must be from preparation stage']

api_operation: Annotated[str, 'API operation to simulate, e.g., "oci:ListBuckets"']

where_context: Annotated[dict[str, str], 'Variable input mapping for required where fields']

checked_statements: NotRequired[Annotated[list[str], 'Statement IDs (UI only, omit for MCP/server)']]

class oci_policy_analysis.application.core.models.models_simulation.SimulationBatchRequest[source]

Bases: TypedDict

Batch simulation request (multiplex multiple scenarios).

Example:

{
  "simulations": [
    { ... see SimulationScenario ... }
  ],
  "trace": true
}

simulations: Annotated[list[SimulationScenario], 'List of simulation scenarios to run']

trace: NotRequired[Annotated[bool, 'If true, include trace output in SimulationResult']]

class oci_policy_analysis.application.core.models.models_simulation.SimulationResult[source]

Bases: TypedDict

Canonical result for a single simulation scenario.

All fields strictly correspond to engine outputs and are fully JSON-serializable.

result: Annotated[Literal['YES', 'NO'], "'YES' if API operation permitted, 'NO' if denied"]

api_call_allowed: Annotated[bool, 'True if API op permitted']

final_permission_set: Annotated[list[str], 'Permissions granted after simulation']

required_permissions_for_api_operation: Annotated[list[str], 'Permissions required for op']

missing_permissions: Annotated[list[str], 'Any permissions missing for full allow']

related_permission_checks: NotRequired[Annotated[list[dict], 'Advisory related resource/permission checks for troubleshooting']]

failure_reason: Annotated[str, 'Reason for denial or error, empty if successful']

scenario_internal_id: NotRequired[Annotated[str, 'Scenario identifier associated with this result']]

scenario_name: NotRequired[Annotated[str, 'Scenario name associated with this result']]

simulation_name: NotRequired[Annotated[str, 'Simulation name associated with this result']]

trace_statements: NotRequired[Annotated[list[dict], 'Statement-by-statement trace, if trace=True']]

class oci_policy_analysis.application.core.models.models_simulation.SimulationBatchResponse[source]

Bases: TypedDict

Batch simulation output: result list, matches input scenario order.

results: Annotated[list[SimulationResult], 'Simulation result(s) for each scenario, in order']

class oci_policy_analysis.application.core.models.models_simulation.ProspectiveStatementInput[source]

Bases: TypedDict

Lightweight input model for defining a prospective (what-if) statement.

These fields mirror the data used by the UI’s prospective editor and are consumed by PolicySimulationEngine.set_prospective_statements.

compartment_path: Annotated[str, 'Effective compartment path, e.g. "ROOT/Finance"']

description: Annotated[str, 'Human-readable label or purpose for this statement']

statement_text: Annotated[str, 'Raw OCI policy statement text']

class oci_policy_analysis.application.core.models.models_simulation.ProspectiveStatementSummary[source]

Bases: TypedDict

Summarized view of a prospective statement for MCP/UI listing.

internal_id: Annotated[str | None, 'Engine-assigned internal id (prospective-*)']

policy_name: Annotated[str | None, 'Display name used for this statement']

compartment_path: Annotated[str | None, 'Effective compartment path for the statement']

parsed: Annotated[bool | None, 'True if parse succeeded']

valid: Annotated[bool | None, 'True if statement is considered valid']

invalid_reasons: Annotated[list[str] | None, 'Reasons for invalid/parse failure if any']

statement_text: Annotated[str | None, 'Original raw policy statement text']

class oci_policy_analysis.application.core.models.models_simulation.ProspectiveStatementResult[source]

Bases: TypedDict

Result of adding or validating a single prospective statement via MCP.

parsed: Annotated[bool, 'True if the statement parsed successfully']

valid: Annotated[bool, 'True if the parsed statement is considered valid']

invalid_reasons: Annotated[list[str], 'Reasons for invalid result, if any']

internal_id: Annotated[str | None, 'Assigned internal id if added to engine; None on failure']

normalized: Annotated[dict, 'Normalized statement payload from the policy normalizer (if valid)']

message: Annotated[str, 'Human-readable summary of the parse/add outcome']

Repositories and parsing

class oci_policy_analysis.application.core.repo.policy_analysis_repository.PolicyAnalysisRepository[source]

Bases: object

This is the main data repository for Policy, Identity, and Compartment data

During initialization, the entire compartment hierarchy and policy tree is loaded into a central JSON dictionary. This central dictionary is then referenced by functions that filter and return a subset of information for display. Parsing, additional analysis, and import/export are made available by additional functions exposed.

Loading of data starts from load_policies_and_compartments, which loads all compartments and policies recursively

Filtering functions return lists of dataclass objects defined in models.py for easy consumption by UI or CLI layers.

See filter_policy_statements for an example of filtering and returning PolicyStatement objects.

enrich_display_structures() → None[source]: Attach parsed display structures to loaded statements and dynamic groups.

reset_state()[source]: Resets all main state variables (lists, dictionaries, flags, clients, IDs, etc.). Call this before any data (re)load operation for a clean repository state.

initialize_client(use_instance_principal: bool, use_resource_principal: bool = False, session_token: str | None = None, recursive: bool = True, profile: str = 'DEFAULT') → bool[source]

Initializes the OCI client to be used for all data operations

Client can be loaded using PROFILE or Instance Principal authentication methods

Parameters:

use_instance_principal – Whether to attempt Instance Principal signer-based authentication
use_resource_principal – Whether to attempt Resource Principal signer-based authentication
recursive – Whether to load tenancy data across all compartments, or simply the root (tenancy) compartment
session – The named OCI Session Token Profile to use - must be present on the file system in the standard OCI location of .oci/config
profile – The named OCI Profile to use - must be present on the file system in the standard OCI location of .oci/config

Returns:

A boolean indicating whether the client was created successfully. False indicates that an unrecoverable issue occurred setting up the client.

check_statement_location_validity(st)[source]

Checks if the compartment location for a statement is valid (exists and is ACTIVE).

Parameters:: st – The policy statement (dict).
Returns:: None if valid; string message if invalid.

load_compartments_only() → bool[source]: Loads only compartments (hierarchy, flat) using OCI Clients.

load_policies_only() → bool[source]: Loads policies/statements only, assuming compartments are already loaded.

get_compartment_path_for_ocid(compartment_ocid: str | None) → str[source]: Resolve a compartment OCID to a hierarchy path string.

load_policies_and_compartments() → bool[source]: Loads both compartments and all policies using OCI Clients. (Convenience function)

reload_compartment_policy_data() → bool[source]

Reload just the policy/compartment/statement data (not IAM), and update the in-memory timestamp. (No cache operations here—see main.py/App for cache update and UI triggers.)

Returns:: True if the reload succeeded, False otherwise.
Return type:: bool

fetch_tenancy_policy_statement_limits()[source]: Fetch two key limits from OCI Limits service (“Identity”): - policies-count (max policies in tenancy) - statements-count (max statements per policy) Uses _api_call_with_logging to time/log the call. Returns a tuple: (policies_count_limit, statements_per_policy_limit) or (None, None) if unavailable or error.

load_complete_identity_domains(load_all_users: bool = True, compartment_domain_search_depth: int = 1) → bool[source]: Loads users, groups, dynamic groups, and domains for all compartments up to the given depth below the root compartment.

filter_policy_statements(filters: PolicySearch, *, statements: list[RegularPolicyStatement] | None = None) → list[RegularPolicyStatement][source]

Filter policy statements by one or more criteria.

Parameters:

filters (PolicySearch) – Dictionary of filter keys and their values (e.g. verb, resource, permission, group, etc).
statements – Optional alternate list of policy-like statements to filter. When provided, this list is filtered instead of the repository’s regular_statements. This is useful for applying the same JSON filter semantics to prospective or simulated statement sets that are not part of the loaded tenancy data.

Returns:

List of statements matching the filter.

Return type:

list[PolicyStatement]

filter_cross_tenancy_policy_statements(alias_filter: list[str]) → list[RegularPolicyStatement][source]

Filter cross-tenancy policy statements containing any provided alias.

Parameters:: alias_filter (list[str]) – List of aliases to look for in statement text.
Returns:: Filtered cross-tenancy policy statements.
Return type:: list[PolicyStatement]

get_users_for_group(group: Group) → list[User][source]

Return all users that belong to the specified exact group. Membership is determined by matching the group name and domain name.

Parameters:: group (Group) – A dictionary with keys: - ‘domain’: str | None - ‘name’: str
Returns:: A list of Users that belong to the specified group. If the group does not exist or has no members, returns an empty list.
Return type:: list[User]

get_groups_for_user(user: User) → list[Group][source]

Return the list of all Groups that a user is a member of

Parameters:: user (User) – The user to find groups for.
Returns:: A list of Groups that the user is a member of.
Return type:: list[Group]

filter_groups(group_filter: GroupSearch) → list[Group][source]: Filter groups based on the provided filter. Public function used by MCP or UI

filter_users(user_filter: UserSearch) → list[User][source]

Filter users based on the provided filter.

This function is used by the MCP interface and the UI.

Parameters:

user_filter (UserSearch) –

A dictionary with optional keys.

domain_name (list[str]): Domain names to filter by (case-insensitive).
search (list[str]): Search terms to match against usernames and display names (case-insensitive).
user_ocid (list[str]): User OCIDs to filter by (case-insensitive).

Returns:

Users that match the filter criteria.

Each User is represented as a dictionary with keys:

domain_name (str | None): Domain name of the user.
user_name (str): Username.
user_ocid (str): OCID of the user.
display_name (str): Display name of the user.
email (str): Email of the user.
user_id (str): Internal ID of the user.
groups (list[str]): Group OCIDs the user belongs to.

Return type:

list[User]

filter_dynamic_groups(filters: DynamicGroupSearch) → list[DynamicGroup][source]

Filter dynamic groups using JSON-based filters.

Parameters:

filters (DynamicGroupSearch) –

A mapping of filter keys to one or more values.

OR: multiple values within a field act as logical OR.
AND: multiple fields are combined as logical AND.

Supported keys: * domain_name → matches “Domain” * dynamic_group_name → matches “DG Name” * matching_rule → matches “Matching Rule” * dynamic_group_ocid → matches “DG OCID” * in_use → matches “In Use” (True/False)

Returns:

A list of dynamic groups that satisfy the filters.

Each dynamic group is represented as a dictionary with keys: * domain_name (str | None): The domain name of the dynamic group. * dynamic_group_name (str): The name of the dynamic group. * dynamic_group_id (str): The ID of the dynamic group. * dynamic_group_ocid (str): The OCID of the dynamic group. * matching_rule (str): The matching rule of the dynamic group. * description (str): The description of the dynamic group. * in_use (bool): Whether the dynamic group is in use. * creation_time (str): The creation timestamp of the dynamic group. * created_by_name (str): The name of the user who created the dynamic group. * created_by_ocid (str): The OCID of the user who created the dynamic group.

Return type:

list[DynamicGroup]

Raises:

ValueError – If an unknown filter key is provided.

load_from_compliance_output_dir(dir_path: str, load_all_users: bool = True) → bool[source]

Load all compartments, domains, groups, users, dynamic groups, and policies from compliance tool output files.

Always resets the reload time (policy_data_reloaded) so that reload is not shown for compliance/CSV data.

Starts with domains, then dynamic groups, then users/groups/membership, then compartments, then policies. This function is for offline/compliance output analysis: no attempt to initialize any OCI client.

Parameters:

dir_path (str) – Path to a directory containing the expected compliance output files.
load_all_users (bool) – If False, skip loading users. Default is True.

Returns:

True if all files parsed and data loaded successfully, False otherwise.

Return type:

bool

class oci_policy_analysis.application.core.repo.reference_data_repo.ReferenceDataRepo(json_dir: str | None = None)[source]

Bases: object

Load OCI resource, permission, family, and operation reference data.

load_data()[source]

get_permission_risk(permission: str, resource: str | None = None)[source]

Get the risk score for a single permission string (optionally for a given resource). If resource is not provided, search all resources.

This method is case-insensitive for permission and resource.

get_permissions_risk_sum(permissions, resource: str | None = None)[source]: Given a list of permissions (case-insensitive), compute the summed risk score.

get_verb_resource_risk(verb: str, resource: str)[source]: Get cumulative permission risk for all permissions associated with given verb/resource.

get_permissions(entity, verb, action='allow')[source]

Get cumulative permissions for a resource or family at a given verb level and action. For “allow”: behavior is as before. For “deny”: logic is inverted – broader verbs (like ‘inspect’) deny more permissions.

Special case: if entity == ‘all-resources’, gather permissions from all resources/types, but for the specified verb only. For ‘allow’, union the specific-verb permissions from all. For ‘deny’, union the same but these are what is DENIED.

Parameters:

entity (str) – Resource name or family name (case-insensitive).
verb (str) – Verb level (‘inspect’, ‘read’, ‘use’, ‘manage’), case-insensitive.
action (str) – “allow” or “deny” (default: “allow”)

Returns:

List of cumulative permissions, always UPPERCASE.

Return type:

list

check_overlap(perm_set1, perm_set2)[source]: Check for overlapping permissions between two permission sets. Uses 2 lists of permissions. Always compares and returns upper case permissions (display, logic, and reporting).

check_overlap_params(entity1, verb1, action1, entity2, verb2, action2)[source]

Check overlapped permissions by specifying both sides as entity/verb/action.

Parameters:

entity1 (str), verb1 (str), action1 (str)
entity2 (str), verb2 (str), action2 (str)

Returns:

List of overlapping permissions.

Return type:

list

get_source(entity)[source]: Retrieve the source URL(s) for a given entity (resource or family) in a case-insensitive manner.

get_containing_family(resource_name: str) → str | None[source]

Return the family that contains the given resource, if any.

The lookup is case-insensitive for the resource name and returns the canonical family key as loaded from reference data JSON files.

Parameters:: resource_name – OCI resource token (for example: subnets).
Returns:: virtual-network-family) if found, otherwise None.
Return type:: The family name (for example

has_api_operation_permissions(operation_name, granted_permissions)[source]

Check if all required permissions for the given API operation are present in the granted_permissions list. :param operation_name: Name of the API operation (as in ‘operations’ node). :type operation_name: str :param granted_permissions: List of permission strings to check. :type granted_permissions: list[str]

Returns:: True if all required permissions for the operation are present, False otherwise.
Return type:: bool

policy_statement_normalizer.py

Centralizes post-parsing normalization of OCI IAM policy statements. - Calls an internal ANTLR parser (PolicyStatementParser) for lexical/syntactical analysis. - Normalizes parsed output to conform to models: DefineStatement, AdmitStatement, EndorseStatement, RegularPolicyStatement. - Handles field splitting, comment extraction, tenancy/principal mapping, etc. - Provides logging for parse/normalization errors and field extraction issues.

Usage:: from .policy_statement_normalizer import PolicyStatementNormalizer stmt = PolicyStatementNormalizer(logger=your_logger).normalize(statement_text, statement_type, base_fields)

class oci_policy_analysis.application.core.parser.policy_statement_normalizer.LoggingErrorListener(logger=None, context_text=None)[source]

Bases: ErrorListener

Custom ANTLR error listener that logs all syntax errors/warnings via the provided logger.

syntaxError(recognizer, offendingSymbol, line, column, msg, e)[source]

reportAmbiguity(recognizer, dfa, startIndex, stopIndex, exact, ambigAlts, configs)[source]

reportAttemptingFullContext(recognizer, dfa, startIndex, stopIndex, conflictingAlts, configs)[source]

reportContextSensitivity(recognizer, dfa, startIndex, stopIndex, prediction, configs)[source]

class oci_policy_analysis.application.core.parser.policy_statement_normalizer.PolicyStatementParser[source]

Bases: object

Parse OCI policy text with the generated ANTLR grammar.

parse(text)[source]

oci_policy_analysis.application.core.parser.policy_statement_normalizer.strip_quotes(val)[source]: Remove one matching pair of single or double quotes from a value.

class oci_policy_analysis.application.core.parser.policy_statement_normalizer.PolicyStatementNormalizer[source]

Bases: object

Normalize parsed OCI policy statements into application models.

normalize(statement_text: str, statement_type: str, base_fields: dict)[source]: Normalize a statement of a given type (define/admit/endorse/regular). base_fields must include all fields from BasePolicyStatement required by models.

OCI Policy Subject Parsing Utility

Centralizes all parsing of subject logic (group, dynamic-group, service, resource, any-user, any-group) for both regular/policy and cross-tenancy statements.

Input: subject_type (str), raw_subject (str)
Output: canonical list of tuples/strings as required by normalization

Author: Plan-based refactor by Cline (2026)

oci_policy_analysis.application.core.parser.policy_subject_parser.parse_policy_subjects(subject_type: str, raw_subject: str)[source]

Parse the raw subject string from ANTLR into canonical representations used in normalization.

Returns:

list of tuples (domain, name), or OCID strings - Service: list of strings - Any-user / any-group: list with single special string - Resource: list of resource names or OCIDs (pass-through) - group-id/dynamic-group-id: list of tuples (None, OCID)

Return type:

Group / dynamic-group

Display-oriented condition structure extraction.

oci_policy_analysis.application.core.parser.condition_structure.parse_condition_structure(text: str | None) → dict[str, Any][source]: Parse a where clause or dynamic group rule into display metadata.

oci_policy_analysis.application.core.parser.condition_structure.format_condition_structure_summary(structure: dict[str, Any] | None) → str[source]: Return compact display text for a parsed structure.

Engines and services

class oci_policy_analysis.application.core.engine.policy_intelligence_engine.PolicyIntelligenceEngine(policy_repo, strategies: list[IntelligenceStrategy] | None = None)[source]

Bases: object

Provides post-load intelligence, overlap analysis, and advanced policy insights on OCI policies.

This engine operates on a loaded PolicyAnalysisRepository and delivers advanced analytics such as risk scores, overlaps, recommendations, and policy hygiene findings.

Parameters:: policy_repo (PolicyAnalysisRepository) – The repository containing loaded compartment, policy, and identity data.

policy_repo

Source of OCI policy, identity, and compartment data.

Type:: PolicyAnalysisRepository

overlay

Stores the full results of all analytics (risk, overlaps, recommendations, etc).

Type:: PolicyIntelligence

permissions_report

Holds the effective permissions report used by various UI components.

Type:: dict

static calculate_principal_key(subject_type: str, domain: str | None, name: str) → str[source]

Return the canonical principal key string for a subject triple.

Delegates to shared helper in policy_helpers for consistent normalization across repo/intelligence/simulation layers.

register_strategy(strategy: IntelligenceStrategy) → None[source]: Register a single intelligence strategy (pluggable).

get_strategy_ids() → list[str][source]: Return strategy_ids in run order (for Settings and run_all filtering).

get_strategies_for_settings() → list[tuple[str, str, str]][source]: Return (strategy_id, display_name, category) for all registered strategies, in run order.

run_all(enabled_strategy_ids: list[str] | None = None, params: dict | None = None) → None[source]

Run all enabled intelligence strategies in order and merge results into overlay.

Ensures prerequisites (compartment index, invalid statements, DG in-use) are run first. If no strategies are registered, falls back to legacy method calls for backward compatibility.

Parameters:

enabled_strategy_ids – If None, run all registered strategies (or legacy path). Otherwise run only these.
params – Optional dict (e.g. where_clause_reduction_pct, service_principal_reduction_pct, enabled_cleanup_check_ids, consolidation_strategy_names). Passed to each strategy; engine ref added.

resolve_cross_tenancy_aliases() → None[source]

Resolve referenced cross-tenancy aliases to define OCIDs.

Parse-time sets alias placeholders as unresolved. This step resolves them after all define/admit/endorse parsing is complete.

build_permissions_report()[source]

Build a detailed nested report of effective permissions for all policy statements.

Scans all statements, resolves permissions and subjects, and groups by effective path and subject. Stores both the structured report and supporting lookup maps in self.permissions_report.

Returns:: A structure with keys “report”, “resource_map”, “perm_conditionals”, and “perm_statements” containing all effective allow/deny permissions and metadata, or empty dicts if no data is present.
Return type:: dict

run_dg_in_use_analysis()[source]: Analyzes Dynamic Group data for unused Dynamic Groups. Should be called after repo is loaded and statements parsed.

analyze_policy_overlap()[source]: Analyze policy statements for potential overlaps, calling after all statements loaded/parsed. Stores result in self.overlay[“overlaps”] using the new PolicyIntelligence overlay structure.

get_policy_overlaps_by_internal_id(internal_id: str)[source]: Returns all PolicyOverlap entries for a given policy statement internal ID by looking up the overlay model.

find_invalid_statements()[source]

Mark regular policy statements as invalid if they fail various validity checks, such as: - Nonexistent Dynamic Groups or Groups - Tag-based where-clause references to missing defined tag namespace/key - Invalid compartment OCIDs - Invalid verbs/resources

This method modifies the statements in-place, adding an invalid_reasons list if applicable.

calculate_effective_compartment_for_statement(st)[source]: Calculate effective compartment OCID and path for a single statement, mutating st in place. Uses indexes built via build_compartment_index().

calculate_all_effective_compartments()[source]: Resolve effective compartment for all statements. Loop through all statements and calculate.

build_compartment_index()[source]: Build quick-lookup structures for resolving compartment names and parent/child relationships used by calculate_effective_compartment_for_statement().

calculate_potential_risk_scores(where_clause_reduction_pct=50, service_principal_reduction_pct=50)[source]

Calculates potential risk scores for each policy statement.

Formula: risk = exposure_points × compartments_in_scope (then optional reductions for WHERE clause or service principal). Exposure points = sum of each permission’s risk by verb level from the reference data (inspect=1, read=5, use=50, manage=100). When the statement has a permission list, each permission is looked up and its verb-level risk is summed. When it has no permission list, the reference repo returns the sum of all permissions for that verb/resource (each weighted by its verb). Compartments in scope = number of compartments at or below the statement’s effective path in the hierarchy (e.g. “in tenancy” with 12 compartments => multiplier 12; a statement effective in a sub-compartment has a smaller multiplier).

Applies a raw score reduction for statements with a WHERE clause, and a further reduction for service-principal statements with use/manage verbs (adjustable via service_principal_reduction_pct).

Results are stored in self.overlay[“risk_scores”] as a list of dicts:: { “statement_internal_id”, “score”, “notes”, “recommendations” }

build_cleanup_items(enabled_check_ids=None)[source]

Analyze policy repository and collect actionable cleanup items for all key risk categories. This should be called BEFORE build_overall_recommendations.

Parameters:: enabled_check_ids – If None, all checks run. Otherwise only run checks whose ID is in this list (use CLEANUP_CHECK_IDS). Disabled keys are set to empty list in overlay.

The lists are attached to self.overlay[“cleanup_items”].

build_overall_recommendations(params: dict | None = None)[source]

Build the overall (user-facing) recommendations list for overlay[“recommendations”]. Each recommendation summarizes the count of existing actionable issues. Assumes build_cleanup_items has already been called and populated “cleanup_items”. If no real recommendations are found, yields one informational finding as a placeholder. params may include consolidation_strategy_names (list of display names) for workbench cross-link.

Example of recommendation dict:

{
    'Recommendation': 'Investigate invalid policy statements',
    'Priority': 'High',
    'Category': 'Policy Hygiene',
    'Notes': '3 invalid policy statement(s) detected. Review the cleanup/fix tab for details.',
    'Action': 'Plan: Review and remediate invalid policy statements',
    'ActionDetail': 'Examine policies with invalid statements and resolve as appropriate.',
}

This could change in the future to include risk score-based recommendations.

build_policy_consolidation()[source]

Analyze policies/statements for possible consolidation opportunities and populate overlay[“consolidations”] with a list of dicts:

{
    "Statement": ...,
    "Policy Name(s)": ...,
    "Compartment": ...,
    "Principal": ...,
    "Service/Resource": ...,
    "Consolidation Reason": ...,
    "Action": ...,         # always present now
    "ActionDetail": ...    # optional, for detail/planning dialog
}

Criteria:

Policies with only a single statement (likely consolidation candidate).
Statements with identical principal, compartment, and service/resource, but split across multiple differently-named policies (should suggest merge).

class oci_policy_analysis.application.core.engine.policy_simulation_engine.PolicySimulationEngine(policy_repo=None, ref_data_repo=None)[source]

Bases: object

Central simulation service for OCI policies.

— Stateless simulation setup APIs (recommended for UI/MCP) —

These methods provide the canonical way to derive principal keys, resolve all applicable policy statements, and compute required where-clause parameter names, regardless of UI vs MCP flow.

Step 1: Normalize compartment_path and principal_type/principal into principal_key (handles tuple/string, None/’default’, etc).: -> Use normalize_principal_key(principal_type, principal)
Step 2: Find all statements for a context (compartment_path, principal_type, principal).: -> Use find_applicable_statements_for_context(compartment_path, principal_type, principal)
Step 3: Get all required where-clause variable names for a context.: -> Use get_required_where_fields_for_context(compartment_path, principal_type, principal)

The MCP version always uses all valid applicable statements for a context.

Responsibilities: - Given principal + effective compartment, generate the list of all applicable policy statements (including ‘any-user’). - Build and cache a mapping of compartment -> principal -> [statements] for fast lookup. - For a given policy context, extract all required where-clause fields, and evaluate conditional logic against input values. - For a set of user choices (principal + compartment + where values), return the effective permission set and permission simulation trace. - Fold in/replace all old where simulation logic found elsewhere (single source of truth). - DO NOT load or store raw data — consumption only.

Primary entrypoints:

get_applicable_statements(principal, compartment): List[dict]
get_required_where_fields(statements): Set[str]
simulate_permissions(principal, compartment, where_context: dict): (Set[str], List[dict])
simulate_api_operation(principal, compartment, operation, where_context: dict): dict (YES/NO + full trace)

get_statements_for_context(compartment_path, principal_type, principal=None)[source]

Canonical: Return (principal_key, [statement dicts]) for a given policy simulation context (UI/MCP).

Parameters:

compartment_path (str) – Path like ‘ROOT/Finance’
principal_type (str) – “user”, “group”, etc.
principal (optional) – None, string, or (domain, name)

Returns:

Tuple with normalized principal key and list of matching statements.

Return type:

(principal_key, statements)

See docs/source/simulation_engine_usage_examples.py for usage.

get_required_where_fields_for_context(compartment_path, principal_type, principal=None)[source]

Canonical: Given a context, return (principal_key, set of required where input variables).

Parameters:

compartment_path (str) – e.g. “ROOT/Finance”
principal_type (str) – “user”, “group”, etc.
principal (optional) – None, string, or (domain, name)

Returns:

Tuple with normalized principal key and set of all required where fields.

Return type:

(principal_key, required_fields)

See docs/source/simulation_engine_usage_examples.py for usage.

static extract_variable_names(cond_str)[source]

Extract variable names from a policy condition string using the OCI IAM Policy Condition parser.

Parameters:: cond_str (str) – The condition string to analyze.
Returns:: A set of variable names (as strings) present in the condition clause.
Return type:: set[str]

get_prospective_statements() → list[dict[str, Any]][source]

Return a shallow copy of all current prospective statements.

These statements are in addition to the regular policy statements loaded from the tenancy/cache and are used for “what-if” simulation scenarios. They are not written back to OCI.

set_prospective_statements(statements: list[dict[str, Any]]) → None[source]

Replace the current set of prospective statements.

The caller (UI/MCP) provides lightweight dicts with at least compartment_path, statement_text, and optional description. This method will:

mark each as is_prospective = True
ensure each has a unique internal_id

Persistence to disk/settings is orchestrated by the application; the engine simply owns the in-memory representation used for simulation.

validate_prospective_statement(statement_text: str) → dict[str, Any][source]

Validate and normalize a single prospective statement.

Returns a dict with keys:

parsed: bool
valid: bool
invalid_reasons: list[str]
normalized: dict (if valid)

simulate_and_record(principal_key: str, effective_path: str, api_operation: str, where_context: dict, checked_statement_ids: list[str] | None = None, trace_name: str | None = None, scenario_internal_id: str | None = None, scenario_name: str | None = None, simulation_name: str | None = None, trace: bool = False) → dict[str, Any][source]

Simulates permissions for a principal, compartment, and operation, recording full trace/results.

Parameters:

principal_key (str) – The unique principal key (e.g., “group:default/Admins”) for the simulation.
effective_path (str) – The effective compartment path for simulation.
api_operation (str) – The API operation to check permission for.
where_context (dict) – Input/context for any evaluated ‘where’ policy clauses.
checked_statement_ids (Optional[list[str]]) – List of policy statement internal_ids to include in simulation. If omitted or None, all applicable statements for the context (principal_key and effective_path) are used.
trace_name (str, optional) – Optional name for this simulation trace/history.
trace (bool, optional) – If True, includes detailed step-by-step trace; else, summary only.

Returns:

Full simulation result containing the final decision, missing permissions, required permissions, failure reason, and a complete simulation trace. The UI controls how much trace detail it displays; history always retains the complete trace.

Return type:

dict[str, Any]

Example

result = engine.simulate_and_record(‘group:default/Admins’, ‘ROOT’, ‘oci:ListBuckets’, {}, checked_ids, trace=True) result = engine.simulate_and_record(‘group:default/Admins’, ‘ROOT’, ‘oci:ListBuckets’, {}, trace=True) # auto selects statements

get_simulation_trace_list() → list[dict[str, str | bool]][source]

Get the list of names and timestamps for all recorded simulation traces.

Returns:

List of dictionaries with keys ‘name’ and ‘timestamp’: corresponding to each simulation history entry.

Return type:

list[dict[str, str]]

Example

[{‘name’: ‘Simulation 1’, ‘timestamp’: ‘2026-01-05T14:00:01’}, …]

get_simulation_trace_by_index(idx: int) → dict[str, Any] | None[source]

Retrieve a previously recorded simulation trace result by history index.

Parameters:: idx (int) – Index into simulation history (0 = oldest).
Returns:: The trace result dict, or None if index is out of range.
Return type:: dict[str, Any] or None

Example

>>> engine.get_simulation_trace_by_index(0)
{'result': 'YES', ...}

get_simulation_trace_by_name(name: str) → dict[str, Any] | None[source]

Retrieve a simulation trace result by name label.

Parameters:: name (str) – Name of the simulation trace to look up.
Returns:: The simulation trace result, or None if no match.
Return type:: dict[str, Any] or None

get_api_operations(filter_text: str = '') → list[source]

Get all known API operation names, optionally filtered by substring.

Parameters:: filter_text (str, optional) – Substring to filter operation names (case-insensitive). Defaults to ‘’ (all ops).
Returns:: Sorted list of API operation names.
Return type:: list[str]

Example

>>> engine.get_api_operations('bucket')
['oci:ListBuckets', 'oci:CreateBucket', ...]

get_applicable_statements(principal_key: str, effective_path: str) → list[dict][source]

Get all applicable policy statements for a principal and compartment by building a proper PolicySearch filter using the principal_key and effective_path.

Parameters:

principal_key (str) – Canonical principal key (e.g., ‘user:Default/anita’)
effective_path (str) – Compartment path for simulation context (e.g., ‘ROOT/Finance’)

Returns:

List of matching policy statement dicts.

Return type:

List[dict]

Notes

Used by all canonical orchestration flows (UI/MCP)
See usage examples in docs/source/simulation_engine_usage_examples.py

get_required_where_fields(statements: list[dict]) → set[str][source]

Extract all unique “where clause” input field names needed by a set of policy statements.

Parameters:: statements (list[dict]) – The statements to examine for conditional/where fields.
Returns:: Set of all unique field names required as variables for condition evaluation.
Return type:: set[str]

Example

>>> fields = engine.get_required_where_fields(statements)
>>> print(fields)  # e.g. {'request.time', 'user.department'}

generate_where_clauses_for_statements(statements: list[dict]) → dict[source]

Utility: Generate a mapping from statement internal_id to set of where-variable names required (for UI preview purposes).

Parameters:: statements (list[dict]) – Statements to analyze.
Returns:: {internal_id: set(field names)} for statements that have where conditions.
Return type:: dict

Service facade for policy statement filtering.

class oci_policy_analysis.application.services.analysis_service.FilterResult(total: int, matched: int, statements: list[RegularPolicyStatement])[source]

Bases: object

Result container for statement filtering operations.

total: int

matched: int

statements: list[RegularPolicyStatement]

class oci_policy_analysis.application.services.analysis_service.AnalysisService(context: AppContext)[source]

Bases: object

Expose filtering on loaded policy statements.

filter_policy_statements(filters: PolicySearch) → FilterResult[source]

Filter all loaded policy statements by provided filters.

Parameters:: filters – Search filters applied to policy statements.
Returns:: Total statement count and matched statements.
Return type:: FilterResult

filter_policy_statements_from_payload(payload: dict[str, object]) → FilterResult[source]

Build filters from raw payload and execute statement filtering.

Parameters:: payload – Dictionary payload containing raw filter values.
Returns:: Filtered statement result.
Return type:: FilterResult

filter_policy_statements_subset(*, filters: PolicySearch, statements: list[RegularPolicyStatement]) → FilterResult[source]

Filter a provided subset of statements with policy search criteria.

Parameters:

filters – Search filters applied to the subset.
statements – Preselected statements to search within.

Returns:

Result scoped to the provided statement subset.

Return type:

FilterResult

Load and reload workflows for policy data.

class oci_policy_analysis.application.services.load_service.LoadResult(success: bool, message: str, summary: dict[str, int] | None = None)[source]

Bases: object

Result wrapper for policy load operations.

success: bool

message: str

summary: dict[str, int] | None = None

class oci_policy_analysis.application.services.load_service.LoadService(context: AppContext)[source]

Bases: object

Orchestrate tenancy/cache/compliance loads without UI coupling.

load_from_cache(cache_name: str, *, run_post_load_intelligence: bool = True, post_load_profile: str = 'full', on_stage: Callable[[str, str, str], None] | None = None) → LoadResult[source]

Load repository data from a named cache entry.

Parameters:

cache_name – Cache entry to load.
run_post_load_intelligence – Whether to run post-load processing.
post_load_profile – full for UI overlays/recommendations, minimal for CLI/MCP enrichment, none to skip.
on_stage – Optional stage progress callback.

Returns:

Load status and data summary.

Return type:

LoadResult

load_from_export_json(file_path: str, *, run_post_load_intelligence: bool = True, post_load_profile: str = 'full', on_stage: Callable[[str, str, str], None] | None = None) → LoadResult[source]

Load repository data from an exported JSON file.

Parameters:

file_path – Path to export JSON file.
run_post_load_intelligence – Whether to run post-load processing.
post_load_profile – full for UI overlays/recommendations, minimal for CLI/MCP enrichment, none to skip.
on_stage – Optional stage progress callback.

Returns:

Load status and data summary.

Return type:

LoadResult

load_from_compliance_output(dir_path: str, *, load_all_users: bool = True, run_post_load_intelligence: bool = True, post_load_profile: str = 'full', on_stage: Callable[[str, str, str], None] | None = None) → LoadResult[source]

Load repository data from compliance output directory.

Parameters:

dir_path – Directory containing compliance output artifacts.
load_all_users – Whether to load all users from identity data.
run_post_load_intelligence – Whether to run post-load processing.
post_load_profile – full for UI overlays/recommendations, minimal for CLI/MCP enrichment, none to skip.
on_stage – Optional stage progress callback.

Returns:

Load status and data summary.

Return type:

LoadResult

load_from_tenancy(*, use_instance_principal: bool, use_resource_principal: bool = False, profile: str | None = None, session_token: str | None = None, recursive: bool = False, load_all_users: bool = True, compartment_domain_search_depth: int = 1, run_post_load_intelligence: bool = True, post_load_profile: str = 'full', save_cache_after_load: bool = True, on_stage: Callable[[str, str, str], None] | None = None) → LoadResult[source]

Load policy data directly from OCI tenancy.

Parameters:

use_instance_principal – Whether to authenticate with instance principal.
use_resource_principal – Whether to authenticate with resource principal.
profile – Optional OCI profile name.
session_token – Optional session token for auth.
recursive – Whether tenancy traversal should be recursive.
load_all_users – Whether to load all users for identity domains.
compartment_domain_search_depth – Identity domain search depth.
run_post_load_intelligence – Whether to run post-load processing.
post_load_profile – full for UI overlays/recommendations, minimal for CLI/MCP enrichment, none to skip.
save_cache_after_load – Whether to persist a fresh cache entry after live load.
on_stage – Optional stage progress callback.

Returns:

Load status and data summary.

Return type:

LoadResult

Service facade for policy simulation workflows.

class oci_policy_analysis.application.services.simulation_service.SimulationPrepareResult(principal_key: str, required_where_fields: list[str])[source]

Bases: object

Result payload returned by simulation context preparation.

principal_key: str

required_where_fields: list[str]

class oci_policy_analysis.application.services.simulation_service.SimulationService(context: AppContext)[source]

Bases: object

Expose simulation operations without UI coupling.

prepare(*, compartment_path: str, principal_type: str, principal: object | None) → SimulationPrepareResult[source]

Prepare simulation context for a principal/compartment pairing.

Parameters:

compartment_path – Target compartment path.
principal_type – Principal type value expected by simulation engine.
principal – Principal object used for context resolution.

Returns:

Principal key and required where-fields.

Return type:

SimulationPrepareResult

run(*, principal_key: str, compartment_path: str, api_operation: str, where_context: dict[str, Any], checked_statement_ids: list[str] | None = None) → dict[str, Any][source]

Run a policy simulation and record decision details.

Parameters:

principal_key – Resolved simulation principal key.
compartment_path – Effective path used for simulation.
api_operation – API operation being evaluated.
where_context – Resolved where-clause context values.
checked_statement_ids – Optional set of specific statement IDs to evaluate.

Returns:

Simulation output payload.

Return type:

dict[str, Any]

Service facade for read-only recommendations data shaping.

This service adapts policy intelligence overlay outputs into stable web payloads for the recommendations page. It intentionally excludes mutable workbench actions and focuses on display-only analytics.

class oci_policy_analysis.application.services.recommendations_service.RecommendationsService(context: AppContext)[source]

Bases: object

Provide read-only recommendations datasets for web consumers.

Parameters:: context – Shared application context.

STATEMENT_LIMIT = 500

NEARING_THRESHOLD = 0.85

get_dashboard_payload() → dict[str, Any][source]

Return full recommendations dashboard payload.

Returns:: Read-only recommendations sections for web display.
Return type:: dict[str, Any]

Support

class oci_policy_analysis.application.core.support.caching.CacheManager(cache_dir: Path | None = None)[source]

Bases: object

Handles saving and loading cached JSON data (IAM + AI). Also supports canonical consolidation session persistence: per-tenancy state files for protected_set + plan history. Each cache is tied to a tenancy name and date. The cache directory is ~/.oci-policy-analysis/cache by default, but can be overridden. Caches have the concept of being “preserved” to avoid automatic deletion during culling.

get_or_create_consolidation_state(tenancy_ocid) → dict[source]: Load the canonical consolidation state file for this tenancy_ocid. If it does not exist, create a blank state (protected_set={}, history=[]). Returns the state dict; saves it if it was new.

save_consolidation_state(tenancy_ocid, state: dict)[source]: Overwrite the canonical consolidation state file for this tenancy_ocid.

get_protected_set(tenancy_ocid)[source]: Return the protected_set from consolidation state. Creates file if needed.

set_protected_set(tenancy_ocid, protected_set: dict)[source]: Update and persist the protected_set in the canonical consolidation state file.

get_history(tenancy_ocid)[source]: Return the list of run/plan history for this tenancy_ocid. If no tenancy_ocid, returns empty list (safe for new sessions).

add_run_record(tenancy_ocid, run_record: dict)[source]: Append a new run/plan record to history and save the canonical file.

update_run_record(tenancy_ocid: str, consolidation_effort_id: str, updates: dict) → bool[source]

Update an existing run record in history by consolidation_effort_id.

Parameters:

tenancy_ocid – Tenancy OCID.
consolidation_effort_id – Run identifier (plan_id).
updates – Dict of fields to merge into the run record (e.g. status, step_status).

remove_run_record(tenancy_ocid: str, consolidation_effort_id: str) → bool[source]

Remove a run record from history by consolidation_effort_id.

Parameters:

tenancy_ocid – Tenancy OCID.
consolidation_effort_id – Run identifier (plan_id) to remove.

list_consolidation_tenancy_ocids() → list[str][source]: Return list of tenancy OCIDs that have consolidation state files.

list_all_consolidation_plans() → list[dict][source]: Return a flat list of all consolidation plans across tenancies. Each item has tenancy_ocid, consolidation_effort_id, created_at, and run (full record).

load_prospects(tenancy_ocid: str) → list[dict[str, Any]][source]: Load standalone prospective statements for a tenancy from cache/prospects.

save_prospects(tenancy_ocid: str, rows: list[dict[str, Any]]) → None[source]: Persist standalone prospective statements for a tenancy to cache/prospects.

save_consolidation_session(session_data: dict, plan_id: str | None = None, tenancy_ocid: str | None = None, preserved: bool = False) → str[source]

Save a consolidation session (see models_consolidation.ConsolidationSession) to disk as JSON. Returns file name/path. :param session_data: The dict/TypedDict structure matching ConsolidationSession (pluggable, robust to version drift) :param plan_id: Optional label/ID for file naming, else uses timestamp. :param tenancy_ocid: Optional for nested cache structure. :param preserved: Mark this session as not-to-be-deleted.

Returns:: The output file path string.

load_consolidation_session(plan_id: str, tenancy_ocid: str | None = None) → dict | None[source]

Load a consolidation session from cache. :param plan_id: Label/filename (w/o json) for the session/plan. :param tenancy_ocid: Optional subdir

Returns:: session_data (dict) or None

save_combined_cache(policy_analysis: PolicyAnalysisRepository, export_file=None, preserved: bool = False) → str[source]

Save combined cache for policies and dynamic groups. Returns file name.

Parameters:

export_file – Optional file handle to export to instead of saving to cache directory
preserved – Whether to mark this cache as preserved (not auto-deleted)

Returns:

The name of the file saved

load_combined_cache(policy_analysis: PolicyAnalysisRepository, named_cache: str) → str[source]

Load combined cache for policies and dynamic groups.

Given the name and data of a cache, loads the data into both of the centralized structures for Compartment/Policy JSON storage.

Parameters:: named_cache – The tenancy_date string of the cache name to load
Returns:: A string indicating the name of the file used

load_cache_from_json(policy_analysis: PolicyAnalysisRepository, loaded_json: dict) → bool[source]

Load combined cache data from a given JSON dict. Given loaded JSON data, loads the data into both of the centralized structures for Compartment/Policy JSON storage.

Parameters:: loaded_json – The loaded JSON data as a dict

get_preserved_cache_set() → set[source]: Get a set of cache names which are marked as preserved.

get_available_cache(tenancy_name: str | None) → list[str][source]

Get available cache files for a given profile

If given no argument, simply return the list of all avialable cache files that exist in the cache directory. Entries will contain the tenancy name and date loaded.

Parameters:

tenancy_name – The name of an OCI tenancy, which will filter the cache list down to only
tenancy. (caches for that)

Returns:

a list of the available named caches

load_cache_into_local_json(cached_tenancy: str) → dict[source]

Takes a named cache (tenancy_date) and returns the loaded JSON data as a dict. Used for exporting or other purposes.

Parameters:: cached_tenancy – The tenancy_date string of the cache name to load
Returns:: The loaded cache data as a dict

remove_cache_entry(named_cache: str) → bool[source]

Remove specified cache file AND its entry from cache_entries.json.

Parameters:: named_cache – The tenancy_date string of the cache name to remove
Returns:: True if both file and entry were removed, False otherwise

rename_cache_entry(old_named_cache: str, new_named_cache: str) → bool[source]

Rename both the cache file and its entry in cache_entries.json.

Parameters:

old_named_cache – The current tenancy_date string of the cache name
new_named_cache – The new tenancy_date string of the cache name

Returns:

True if both file and entry were renamed, False otherwise. Returns False (and does NOT rename) if a cache file or entry already exists with the new name.

preserve_cache_entry(named_cache: str, preserve: bool = True) → bool[source]

Mark or unmark a cache entry as preserved in cache_entries.json.

Parameters:

named_cache – The tenancy_date string of the cache name to update
preserve – True to mark as preserved, False to unmark

Returns:

True if the entry was updated, False otherwise

update_policy_section(policy_analysis: PolicyAnalysisRepository, policy_data_reloaded: str | None)[source]

Update ONLY the policies, policy_statements, tag catalog, compartments, defined_aliases, and cross_tenancy_statements in the most recent cache file for a given tenancy, and set ‘policy_data_reloaded’ with the supplied timestamp. This preserves IAM/user/group data and other session metadata. No effect if no cache is present.

Parameters:

policy_analysis – PolicyAnalysisRepository with fresh policy/compartment data in memory
policy_data_reloaded – ISO timestamp string for reloaded policy data

Legacy re-exports for display helpers (compatibility shim).

oci_policy_analysis.application.core.support.helpers.for_display_admit(statement) → dict[source]: Display-friendly dict for parsed AdmitStatement, mapped to UI columns.

oci_policy_analysis.application.core.support.helpers.for_display_define(define: DefineStatement) → dict[source]: Return a dictionary suitable for display purposes for defines.

oci_policy_analysis.application.core.support.helpers.for_display_dynamic_group(dg: DynamicGroup) → dict[source]: Return a dictionary suitable for display purposes for dynamic groups.

oci_policy_analysis.application.core.support.helpers.for_display_endorse(statement) → dict[source]: Display-friendly dict for parsed EndorseStatement, mapped to UI columns.

oci_policy_analysis.application.core.support.helpers.for_display_group(g: Group) → dict[source]: Return a dictionary suitable for display purposes for groups.

oci_policy_analysis.application.core.support.helpers.for_display_policy(statement: RegularPolicyStatement) → dict[source]: Return a dictionary suitable for display purposes for policy statements.

oci_policy_analysis.application.core.support.helpers.for_display_tag_based_policy_row(statement: RegularPolicyStatement) → dict[source]: Return a compact, tag-focused display dict for a regular statement.

oci_policy_analysis.application.core.support.helpers.for_display_user(u: User) → dict[source]: Return a dictionary suitable for display purposes for users.

class oci_policy_analysis.application.core.support.logger.ForceFlushStreamHandler(stream=None)[source]

Bases: StreamHandler

Stream handler that flushes after every log record.

This keeps console and MCP stderr logs visible promptly, including when records are emitted from worker threads.

emit(record)[source]

Emit a record.

If a formatter is specified, it is used to format the record. The record is then written to the stream with a trailing newline. If exception information is present, it is formatted using traceback.print_exception and appended to the stream. If the stream has an ‘encoding’ attribute, it is used to determine how to do the output to the stream.

oci_policy_analysis.application.core.support.logger.get_logger(component: str | None = None) → Logger[source]

Return a component logger. No handlers added (propagate to root).

Parameters:: component – Component name (e.g., ‘cli’, ‘data_repo’). If None, uses base name.
Returns:: Logger instance.

oci_policy_analysis.application.core.support.logger.set_log_level(level: str | int, announce: bool = True) → None[source]

Set root logger level (affects all non-overridden loggers).

Use for global (root) logging threshold, e.g., on app startup or when user changes global log level.
When –verbose is active, will force DEBUG everywhere (but UI always filters DEBUG).
Do not use to configure UI/ConsoleTab handler; that always filters at INFO+.

Options:

CRITICAL 50
ERROR 40
WARNING 30
INFO 20
DEBUG 10

Parameters:

level – Level name (e.g., ‘DEBUG’) or int.
announce – Whether to log the resulting level change.

oci_policy_analysis.application.core.support.logger.set_component_level(component: str, level: str | int) → None[source]

Set level for a specific logger (app or third-party). Component can be full logger name (with dots) or just base name. Example: set_component_level(‘cli’, ‘DEBUG’)

Parameters:

component – Component/logger name (e.g., ‘cli’, ‘data_repo’, ‘requests’)
level – Level name (e.g., ‘DEBUG’) or int.