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.

Runtime context and lifecycle

Construct a context, load an input through LoadService, then query through the focused services below. Full and minimal post-load profiles have different outputs; successful loading is not a claim of complete data or OCI-authoritative access. See Embedding and Integrating OCI Policy Analysis for embedding examples and lifecycle guidance.

Shared application context for orchestrated services.

class oci_policy_analysis.application.context.AppContext(settings: dict[str, ~typing.Any], policy_repo: ~oci_policy_analysis.application.core.repo.policy_analysis_repository.PolicyAnalysisRepository, reference_data: ~oci_policy_analysis.application.core.repo.reference_data_repo.ReferenceDataRepo, intelligence: ~oci_policy_analysis.application.core.engine.policy_intelligence_engine.PolicyIntelligenceEngine, simulation: ~oci_policy_analysis.application.core.engine.policy_simulation_engine.PolicySimulationEngine, cache: ~oci_policy_analysis.application.core.support.caching.CacheManager, cache_service: ~oci_policy_analysis.application.services.cache_service.CacheService, settings_service: ~oci_policy_analysis.application.services.settings_service.SettingsService, reference_data_service: ~oci_policy_analysis.application.services.reference_data_service.ReferenceDataService, logging_service: ~oci_policy_analysis.application.services.logging_service.LoggingService, status: dict[str, ~typing.Any] = <factory>, corpus: ~oci_policy_analysis.application.core.policy_corpus.PolicyCorpus | None = None, cross_tenancy_corpus: ~oci_policy_analysis.application.core.cross_tenancy_corpus.CrossTenancyCorpus | None = None)[source]

Bases: object

Shared mutable runtime state for one active analysis snapshot.

Construct with explicit dependencies or use from_settings for the standard wiring. The factory loads bundled reference data, but does not collect OCI IAM/policies or compile a corpus. Use LoadService before querying the snapshot.

Variables:

settings (dict[str, Any]) – Active configuration dictionary, shared with settings services.
policy_repo (oci_policy_analysis.application.core.repo.policy_analysis_repository.PolicyAnalysisRepository) – Repository retaining source data and derived statement fields.
reference_data (oci_policy_analysis.application.core.repo.reference_data_repo.ReferenceDataRepo) – Bundled resource, permission, and API-operation mappings.
intelligence (oci_policy_analysis.application.core.engine.policy_intelligence_engine.PolicyIntelligenceEngine) – Engine holding validation findings and intelligence overlays.
simulation (oci_policy_analysis.application.core.engine.policy_simulation_engine.PolicySimulationEngine) – Engine for request-context analysis and prospective statements.
cache (oci_policy_analysis.application.core.support.caching.CacheManager) – Persistent repository snapshot manager.
cache_service (oci_policy_analysis.application.services.cache_service.CacheService) – Cache import/export operations over the snapshot manager.
settings_service (oci_policy_analysis.application.services.settings_service.SettingsService) – Settings access and persistence facade.
reference_data_service (oci_policy_analysis.application.services.reference_data_service.ReferenceDataService) – Query facade for bundled reference mappings.
logging_service (oci_policy_analysis.application.services.logging_service.LoggingService) – Global and component logging configuration facade.
status (dict[str, Any]) – Latest progress stage, detail, state, and UTC update timestamp.
corpus (oci_policy_analysis.application.core.policy_corpus.PolicyCorpus | None) – Compiled regular policy facts, or None before enrichment.
cross_tenancy_corpus (oci_policy_analysis.application.core.cross_tenancy_corpus.CrossTenancyCorpus | None) – Cross-tenancy sources/relationships, or None before enrichment.

Notes

Loads replace derived engines and corpora. Consumers should read the current context after each load rather than retaining an old engine or corpus. Context state is not an OCI-authoritative authorization service or an automatic synchronization boundary for concurrent loads and queries.

settings: dict[str, Any]

policy_repo: PolicyAnalysisRepository

reference_data: ReferenceDataRepo

intelligence: PolicyIntelligenceEngine

simulation: PolicySimulationEngine

cache: CacheManager

cache_service: CacheService

settings_service: SettingsService

reference_data_service: ReferenceDataService

logging_service: LoggingService

status: dict[str, Any]

corpus: PolicyCorpus | None = None

cross_tenancy_corpus: CrossTenancyCorpus | None = None

set_status(*, stage: str, detail: str | None = None, state: str = 'idle') → None[source]

Replace the latest progress status for adapters and integrations.

Parameters:

stage – Human-readable operation stage.
detail – Optional progress description; None is stored as an empty string.
state – Adapter-facing state such as idle, running, success, or error.

Notes

Adds a UTC timestamp. This records only the latest status, not an event log.

classmethod from_settings(settings: dict[str, Any], *, verbose_override: bool | None = None) → AppContext[source]

Construct standard repositories, engines, and service wiring.

Parameters:

settings – Active configuration; retained by reference rather than copied.
verbose_override – Explicit runtime logging lock. False applies saved levels exactly; None retains legacy root-DEBUG inference.

Returns:

An unloaded runtime with bundled reference data and no corpus.

Return type:

AppContext

Notes

Applies saved global/component logging levels as a process-wide side effect. Does not authenticate to OCI, load a tenancy/cache, or run post-load analysis.

Shared post-load orchestration helpers for application consumers.

oci_policy_analysis.application.post_load.run_minimal_post_load_enrichment(context: AppContext, *, on_stage: Callable[[str, str, str], None] | None = None) → None[source]

Rebuild the foundations needed for focused snapshot queries.

Parameters:

context – Loaded context to enrich in place.
on_stage – Optional callback receiving stage, detail, and state strings. Callback exceptions propagate to the caller.

Notes

Replaces the intelligence engine, calculates effective paths and validity, marks dynamic-group usage, resolves cross-tenancy aliases, and builds both corpora. Attaches the regular corpus to the existing simulation engine. Does not run the full recommendation strategies or project a permissions report. No OCI refresh occurs here; results reflect the loaded repository. Call through LoadService when progress should also update context.status.

oci_policy_analysis.application.post_load.run_post_load_pipeline(context: AppContext, *, on_stage: Callable[[str, str, str], None] | None = None) → None[source]

Rebuild full intelligence, corpora, reports, and simulation state.

Parameters:

context – Loaded context to enrich in place using its active settings.
on_stage – Optional callback receiving stage, detail, and state strings. Callback exceptions propagate to the caller.

Notes

Replaces the intelligence engine, applies configured strategies and risk parameters, builds cross-tenancy and regular corpora, projects the permissions report, and replaces the simulation engine. Existing simulation/prospective state is not preserved. Some strategies can use available OCI clients. This mutates the context incrementally; failures do not roll back earlier stages. It analyzes the current snapshot, not OCI-authoritative access. Call through LoadService when progress should also update context.status.

Policy corpora

Immutable, valid-only policy knowledge compiled from a loaded snapshot.

The repository deliberately retains invalid and partially understood source statements for diagnostics. PolicyCorpus is the complementary projection: it exposes only the valid, evaluable authorization facts while retaining a record of every excluded source statement.

class oci_policy_analysis.application.core.policy_corpus.CorpusProvenance(policy_name: str, policy_ocid: str, statement_id: str, statement_text: str)[source]

Bases: object

Source evidence attached to an individual permission contribution.

Variables:

policy_name (str) – Display name of the source policy.
policy_ocid (str) – OCI policy identifier when available.
statement_id (str) – Snapshot-local identifier of the source statement.
statement_text (str) – Original policy text, preserving the supplied spelling.

policy_name: str

policy_ocid: str

statement_id: str

statement_text: str

class oci_policy_analysis.application.core.policy_corpus.PolicyCorpusFact(principal_key: str, effective_path: str, grant_path: str, permission: str, effect: Literal['allow', 'deny'], condition: str, provenance: CorpusProvenance, resource: str = '', inherited_from: str = '', original_principal_key: str = '', prospective: bool = False)[source]

Bases: object

One unmerged principal/path/permission contribution in the analysis model.

Variables:

principal_key (str) – Canonical principal identity, normalized on construction.
effective_path (str) – Path represented by this direct or inherited fact.
grant_path (str) – Original path at which the source statement contributes access.
permission (str) – Concrete permission expanded using bundled reference data.
effect (Literal['allow', 'deny']) – Allow or deny contribution; not a final authorization outcome.
condition (str) – Preserved predicate requiring request context if nonempty.
provenance (oci_policy_analysis.application.core.policy_corpus.CorpusProvenance) – Source statement evidence.
resource (str) – Source resource or resource-family name.
inherited_from (str) – Ancestor grant path for inherited facts, otherwise empty.
original_principal_key (str) – Original principal spelling for display.
prospective (bool) – Whether this contribution comes from a hypothetical statement.

principal_key: str

effective_path: str

grant_path: str

permission: str

effect: Literal['allow', 'deny']

condition: str

provenance: CorpusProvenance

resource: str = ''

inherited_from: str = ''

original_principal_key: str = ''

prospective: bool = False

property conditional: bool: Whether this fact requires an OCI where-clause predicate.

property inherited: bool: Whether this effective fact was derived from an ancestor grant.

class oci_policy_analysis.application.core.policy_corpus.CorpusExclusion(statement_id: str, category: str, reason: str, policy_name: str = '', policy_ocid: str = '', statement_text: str = '', action: str = '', resource: str = '', verb: str = '', prospective: bool = False)[source]

Bases: object

Source statement omitted from regular permission-fact compilation.

Variables:

statement_id (str) – Snapshot-local identifier of the omitted statement.
category (str) – Machine-readable exclusion category.
reason (str) – Explanation of the admission failure or unavailable mapping.
policy_name (str) – Source policy display name.
policy_ocid (str) – Source policy OCID when available.
statement_text (str) – Original source text retained for diagnosis.
action (str) – Source effect.
resource (str) – Source resource name.
verb (str) – Source verb.
prospective (bool) – Whether the source was a hypothetical statement.

statement_id: str

category: str

reason: str

policy_name: str = ''

policy_ocid: str = ''

statement_text: str = ''

action: str = ''

resource: str = ''

verb: str = ''

prospective: bool = False

class oci_policy_analysis.application.core.policy_corpus.PolicyCorpus(schema_version: int, built_at: str, facts: tuple[~oci_policy_analysis.application.core.policy_corpus.PolicyCorpusFact, ...]=<factory>, effective_facts: tuple[~oci_policy_analysis.application.core.policy_corpus.PolicyCorpusFact, ...]=<factory>, exclusions: tuple[~oci_policy_analysis.application.core.policy_corpus.CorpusExclusion, ...]=<factory>)[source]

Bases: object

Immutable regular policy facts and transparent compilation diagnostics.

Variables:

schema_version (int) – Version of the compiled representation.
built_at (str) – UTC compilation timestamp, not the OCI collection timestamp.
facts (tuple[oci_policy_analysis.application.core.policy_corpus.PolicyCorpusFact, ...]) – Direct source contributions without descendant expansion.
effective_facts (tuple[oci_policy_analysis.application.core.policy_corpus.PolicyCorpusFact, ...]) – Direct and inherited contributions across known paths.
exclusions (tuple[oci_policy_analysis.application.core.policy_corpus.CorpusExclusion, ...]) – Source statements omitted from concrete permission facts.

Notes

Conditions and denies remain contributions, not resolved access decisions. Coverage depends on the loaded inventory, validation, and permission mappings. Rebuild after replacing or changing repository data.

schema_version: int

built_at: str

facts: tuple[PolicyCorpusFact, ...]

effective_facts: tuple[PolicyCorpusFact, ...]

exclusions: tuple[CorpusExclusion, ...]

to_dict() → dict[str, object][source]: Return a JSON-ready stable representation for adapters and storage.

summary() → dict[str, int][source]: Return compact counts for status surfaces and diagnostics.

inspect_page(section: str, *, page: int = 0, page_size: int = 100, query: str = '') → dict[str, object][source]: Convert only the requested debugger page to JSON-ready records.

class oci_policy_analysis.application.core.policy_corpus.PolicyCorpusBuilder[source]

Bases: object

Compile regular statements into the tool’s representable permission model.

Notes

Explicitly invalid statements and statements missing required fields/mappings are excluded with source evidence. Missing validity flags do not themselves exclude statements. Compilation neither validates through OCI nor evaluates request conditions. Use normal post-load enrichment before compiling.

SCHEMA_VERSION = 1

build(policy_repo: Any, *, additional_statements: list[dict[str, Any]] | None = None) → PolicyCorpus[source]

Compile a detached regular corpus with an optional prospective overlay.

Parameters:

policy_repo – Loaded repository providing normalized regular statements, compartment paths, and optional permission reference data.
additional_statements – Hypothetical statement dictionaries to compile beside the base sources. May include normalized fields under a normalized key.

Returns:

Direct facts, descendant-expanded effective facts, exclusions, schema version, and compilation time. Does not mutate policies in OCI.

Return type:

PolicyCorpus

oci_policy_analysis.application.core.policy_corpus.permissions_report_projection(corpus: PolicyCorpus) → dict[str, object][source]

Project corpus facts into the legacy Permissions Report payload shape.

The report stays a convenient UI projection; it no longer owns a separate parser/permission-expansion/inheritance implementation.

Versioned, immutable snapshot of cross-tenancy source facts and their projection.

class oci_policy_analysis.application.core.cross_tenancy_corpus.CrossTenancyCorpus(schema_version: int, local_tenancy_json: str, definitions_json: tuple[str, ...], statements_json: tuple[str, ...], relationships_json: tuple[str, ...], findings_json: tuple[str, ...], diagnostics_json: tuple[str, ...], projection_json: str)[source]

Bases: object

A snapshot; serialized records are isolated from mutable repository data.

schema_version: int

local_tenancy_json: str

definitions_json: tuple[str, ...]

statements_json: tuple[str, ...]

relationships_json: tuple[str, ...]

findings_json: tuple[str, ...]

diagnostics_json: tuple[str, ...]

projection_json: str

to_dict(section: str | None = None) → dict[str, Any] | list[Any][source]

summary() → dict[str, int][source]

project() → dict[str, Any][source]: Return a detached UI-neutral projection shared by desktop and web.

class oci_policy_analysis.application.core.cross_tenancy_corpus.CrossTenancyCorpusBuilder[source]

Bases: object

Capture all CT source occurrences and their resolved display relationships.

build(repo: Any, findings: list[dict] | None = None) → CrossTenancyCorpus[source]

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.']

policy_ocid: Annotated[list[str], 'Exact, opaque policy identifiers.']

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.']

exact_subject_types: Annotated[list[str], 'Exact case-insensitive parsed subject types; OR across values.']

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.CrossTenancyAssociationEndpoint[source]

Bases: TypedDict

One resource endpoint in a cross-tenancy associate statement.

resource: Annotated[str, 'OCI resource type participating in the association.']

scope: Annotated[str, 'Raw tenancy or compartment scope for the endpoint.']

location_type: Annotated[str, 'Parsed scope kind.']

location: Annotated[str, 'Parsed scope value.']

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.']]

association: NotRequired[Annotated[list[CrossTenancyAssociationEndpoint], 'Ordered association endpoints when action is associate.']]

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], 'Unscoped parsed principal keys; combine with admitted_tenancy to identify a remote principal.']]

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.']]

association: NotRequired[Annotated[list[CrossTenancyAssociationEndpoint], 'Ordered association endpoints when action is associate.']]

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]

supersessions: list[dict]

cross_tenancy_findings: 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.RollbackStep[source]

Bases: TypedDict

Structured inverse operation captured when a consolidation step is planned.

This is intentionally declarative. Renderers can present the inverse now, while a future controlled executor can use the same snapshot without reconstructing state from a changed tenancy.

action: Annotated[Literal['delete_created_policy', 'restore_policy', 'recreate_deleted_policy'], 'Inverse operation for the planned step']

policy_ocid: Annotated[str, 'Original policy OCID, if it still identifies the rollback target']

requires_created_policy_id: NotRequired[Annotated[bool, 'ADD rollback needs the OCI ID created during execution']]

compartment_ocid: NotRequired[Annotated[str, 'Compartment for recreating a deleted policy']]

policy_name: NotRequired[Annotated[str, 'Name for recreating a deleted policy']]

policy_description: NotRequired[Annotated[str, 'Description for recreating a deleted policy']]

statements: NotRequired[Annotated[list[str], 'Policy statements to restore']]

freeform_tags: NotRequired[Annotated[dict[str, str], 'Freeform tags to restore']]

defined_tags: NotRequired[Annotated[dict[str, dict[str, str]], 'Defined tags to restore']]

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']]

rollback: NotRequired[Annotated[RollbackStep, 'Structured inverse operation and immutable rollback snapshot']]

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']

MCP request models that describe policy search inputs to tool consumers.

Bases: BaseModel

One identity selector; populated fields are matched together.

model_config = {'extra': 'allow'}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

principal_type: str | None

domain_name: str | None

name: str | None

ocid: str | None

principal_key: str | None

resource_type: str | None

compartment_ocid: str | None

class oci_policy_analysis.application.core.models.models_mcp.MCPPolicyFilters(*, action: list[Literal['allow', 'deny']] | None = None, principal: MCPPrincipalSelector | None = None, principals: list[MCPPrincipalSelector] | None = None, principal_keys: list[str] | None = None, verb: list[Literal['inspect', 'read', 'use', 'manage']] | None = None, resource: list[str] | None = None, resource_family: str | None = None, permission: list[str] | None = None, policy_name: list[str] | None = None, effective_path: list[str] | None = None, statement_text: list[str] | None = None, **extra_data: Any)[source]

Bases: BaseModel

Discoverable common policy filters, with additional legacy fields retained.

model_config = {'extra': 'allow'}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

action: list[Literal['allow', 'deny']] | None

principal: MCPPrincipalSelector | None

principals: list[MCPPrincipalSelector] | None

principal_keys: list[str] | None

verb: list[Literal['inspect', 'read', 'use', 'manage']] | None

resource: list[str] | None

resource_family: str | None

permission: list[str] | None

policy_name: list[str] | None

effective_path: list[str] | None

statement_text: list[str] | None

class oci_policy_analysis.application.core.models.models_mcp.MCPPolicySearchQuery(*, mode: Literal['simple', 'advanced'] = 'simple', detail_level: Literal['summary', 'simple', 'full'] = 'simple', filters: MCPPolicyFilters | None = None, limit: int = 50, **extra_data: Any)[source]

Bases: BaseModel

The same query fields used by policy_search, usable within a search set.

model_config = {'extra': 'allow'}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

mode: Literal['simple', 'advanced']

detail_level: Literal['summary', 'simple', 'full']

filters: MCPPolicyFilters | None

limit: int

class oci_policy_analysis.application.core.models.models_mcp.MCPPolicySetSearch(*, mode: Literal['simple', 'advanced'] = 'simple', detail_level: Literal['summary', 'simple', 'full'] = 'simple', filters: MCPPolicyFilters | None = None, limit: int = 50, search_id: str = '', label: str = '', required: bool = True, query: MCPPolicySearchQuery | None = None, **extra_data: Any)[source]

Bases: MCPPolicySearchQuery

One named check; direct query fields are preferred, legacy query wrappers remain accepted.

search_id: str

label: str

required: bool

query: MCPPolicySearchQuery | None

model_config = {'extra': 'allow'}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class oci_policy_analysis.application.core.models.models_mcp.MCPPolicySetEvaluation(*, min_confidence: Literal['exact', 'subject_match_with_conditions', 'identity_match_with_residual', 'rule_evidence', 'broad', 'ambiguous'] = 'broad', require_human_principal_coverage: bool = False, require_workload_principal_coverage: bool = False, require_service_principal_coverage: bool = False, require_tag_condition_coverage: bool = False, **extra_data: Any)[source]

Bases: BaseModel

Optional evidence-coverage checks; matching statements are not effective authorization proof.

model_config = {'extra': 'allow'}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

min_confidence: Literal['exact', 'subject_match_with_conditions', 'identity_match_with_residual', 'rule_evidence', 'broad', 'ambiguous']

require_human_principal_coverage: bool

require_workload_principal_coverage: bool

require_service_principal_coverage: bool

require_tag_condition_coverage: bool

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(*, policy_load_method: str | None = None) → bool[source]

Load policies through a selected discovery strategy and shared parsers.

Parameters:: policy_load_method – Optional resource_search (default) or compartment override. Otherwise reuse the current live-load method, then the application setting, then Resource Search.
Returns:: Whether collection completed with usable or legitimately empty discovery results. Individual API failures remain recorded; success does not establish complete tenancy coverage.
Return type:: bool

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]

Refresh tenancy policy limits from every Identity Limits API page.

Returns:: Policy-object and compartment-hierarchy statement limits. Missing values remain unknown; no default is used.
Return type:: tuple[int | None, int | None]

Notes

Requires an initialized Limits client and tenancy OCID. Replaces both in-memory limit representations, logs API diagnostics, and returns missing values for unavailable data or API errors. No cache is written.

set_user_supplied_tenancy_policy_limits(policies_count, chain_count) → dict[str, Any][source]: Set validated limits for only the active tenancy snapshot/cache.

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 a CIS Compliance export from CSV files.

Compartments and policies are required. Identity-oriented artifacts (domains, dynamic groups, groups/membership, and users) are optional, so a policy-and-compartments-only export remains useful for hierarchy, policy movement, consolidation, and policy-derived principal searches.

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

Live policy discovery strategies and their shared detail/parsing pipeline.

oci_policy_analysis.application.core.repo.policy_loading.validate_policy_load_method(value: str) → str[source]

Validate a policy discovery method without silently changing load scope.

Parameters:: value – resource_search or compartment.
Returns:: The validated method.
Return type:: str
Raises:: ValueError – If the method is unsupported.

class oci_policy_analysis.application.core.repo.policy_loading.LivePolicyLoader(repo: PolicyAnalysisRepository)[source]

Bases: object

Discover policy IDs, fetch full details, then normalize on the calling thread.

Parameters:: repo – Initialized repository with its current compartment and IAM inventory.
Variables:: repo – Repository receiving source policies and parsed statement evidence.

Notes

API calls use the repository’s in-memory diagnostics wrapper. Discovery and detail fetches paginate/use bounded workers; parser mutations remain serial. Individual API failures retain partial evidence and collection errors. Total discovery/detail failure raises instead of accepting an apparently empty snapshot. No cache files are written by this loader.

load(method: str) → None[source]

Populate repository policy evidence using one explicit discovery method.

Parameters:

method – Validated discovery method; no automatic fallback is performed.

Raises:

ValueError – If the discovery method is unsupported.
RuntimeError – If all compartment listings or all policy detail calls fail.
Exception – For a failed tenancy-wide Resource Search request.

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]

rebuild_name_maps()[source]: Resolve declared aliases only; singular/plural spelling is not inferred.

resolve_operation(operation_name)[source]: Return a qualified identity, accepting an unambiguous legacy name.

get_catalog_metadata(entity)[source]

Return catalog provenance and explicit coverage limitations.

Parameters:: entity – Resource or family identifier, including a declared alias.
Returns:: Available status, notes, source and discovery metadata. Search terms are descriptive vocabulary, not accepted policy identifiers. Missing metadata does not establish deprecation or complete coverage.
Return type:: dict

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

get_resource_mapping_status(entity)[source]: Return optional mapping-status metadata for a resource token.

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

Notes

Context-dependent or unavailable mappings cannot establish coverage, even when their fixed permission list is empty. Explicitly public operations with an empty list retain their existing behavior.

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.

Variables:

policy_repo (PolicyAnalysisRepository) – Source of OCI policy, identity, and compartment data.
overlay (PolicyIntelligence) – Stores the full results of all analytics (risk, overlaps, recommendations, etc).
permissions_report (dict) – Holds the effective permissions report used by various UI components.

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, on_stage: Callable[[str], None] | 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(corpus: PolicyCorpus | None = None)[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.

get_policy_supersession_by_internal_id(internal_id: str) → dict | None[source]

Return complete-supersession evidence for one statement.

Parameters:: internal_id – Internal ID of the lower-scope candidate statement.
Returns:: The directed supersession finding, or None when the statement is not completely covered by eligible ancestor evidence.

find_invalid_statements()[source]

Mark regular policy statements as invalid if they fail various validity checks, such as: - Nonexistent Dynamic Groups or Groups when their inventories were loaded - 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, corpus: PolicyCorpus | None = 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:

Normalized principal key and matching statement records.

Return type:

tuple[str, list[dict]]

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:

Normalized principal key and required condition-variable names.

Return type:

tuple[str, set[str]]

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_corpus(corpus: PolicyCorpus | None) → None[source]: Attach the compiled base corpus and invalidate its prospective overlay.

get_corpus(*, include_prospective: bool = False) → PolicyCorpus | None[source]: Return the compiled corpus, optionally with a cached prospective overlay.

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

Statement-search result over the loaded repository.

Variables:

total (int) – Number of source statements in the searched repository or subset.
matched (int) – Number of statements satisfying the filters.
statements (list[oci_policy_analysis.application.core.models.models_policy.RegularPolicyStatement]) – Matching normalized source statement records, including invalid records when permitted by the filters.

total: int

matched: int

statements: list[RegularPolicyStatement]

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

Bases: object

Filter source statements without requiring a presentation adapter.

Parameters:

context – Runtime containing the active policy repository.

Variables:

context – Shared runtime queried on each operation.
logger – Component logger for search diagnostics.

Notes

Queries source statements rather than only admitted corpus facts. Empty results do not establish absence of OCI access in an incomplete snapshot.

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

Outcome of a repository load and its selected enrichment profile.

Variables:

success (bool) – Whether the load operation reported success; not a completeness claim.
message (str) – Human-readable outcome or failure description.
summary (dict[str, int] | None) – Entity counts when available, otherwise None. Counts do not establish complete IAM or compartment coverage.

success: bool

message: str

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

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

Bases: object

Replace and enrich the active snapshot without presentation coupling.

Parameters:

context – Runtime containing the repositories and services to update.

Variables:

context – Shared runtime mutated by load operations.
logger – Component logger for loading and progress diagnostics.

Notes

Loads reset repository/derived state and discard prospective state. Failures are not transactional and may leave an empty or partially populated context. Methods return LoadResult for handled load failures; errors from enrichment or caller callbacks can propagate.

The full profile runs intelligence, compiles corpora, builds the permissions report, and replaces simulation. Minimal compiles query foundations/corpora without full recommendations. None skips enrichment. Unknown profile names fall back to full. Setting run_post_load_intelligence=False also skips it. A successful load does not guarantee complete or current OCI data.

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, policy_load_method: str | None = None, 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.
policy_load_method – Optional resource_search or compartment discovery override; otherwise use the saved application setting with Resource Search as the default.
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

Notes

Instance principal loads first probe OCI metadata with 0.5-second connect/read timeouts and no retries. An unavailable endpoint returns a failed result before resetting the snapshot or creating the signer. A successful probe does not validate the instance principal or IAM.

Attempts to retrieve policy limits after policy collection, before enrichment and cache saving, for every post-load profile. Unavailable limits produce a warning without failing the policy load. Offline load methods do not make this request.

Raises:: ValueError – If the policy discovery method is unsupported. Validation happens before resetting the current snapshot.

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

Prepared principal identity and request fields needed for simulation.

Variables:

principal_key (str) – Resolved canonical identity used by the simulation engine.
required_where_fields (list[str]) – Sorted condition-variable names to supply in request context.

principal_key: str

required_where_fields: list[str]

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

Bases: object

Prepare and evaluate requests against the active analysis model.

Parameters:

context – Enriched runtime containing the simulation engine.

Variables:

context – Shared runtime whose current simulation engine is used per call.
logger – Component logger for preparation and execution diagnostics.

Notes

Preparation and execution may update engine state and record results. Supply request-context values for applicable conditions. Results describe the tool’s model of the snapshot, not decisions by OCI’s policy engine.

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

List known environment choices without preparing or evaluating policies.

Returns:: Compartment paths and principal choices grouped by type. Each choice includes a display label, engine input and canonical key. Inventory and retained source references are combined; their presence does not establish evaluation eligibility or complete coverage.
Return type:: dict[str, Any]

Notes

Reads the current snapshot and prospective records without modifying engines, history, source evidence or cache files. Users are offered only when user inventory was collected.

statement_candidates(*, compartment_path: str, principal_type: str, principal: object | None) → dict[str, Any][source]

Discover applicable source evidence and corpus-backed selection eligibility.

Parameters:

compartment_path – Effective compartment selected for the request.
principal_type – Engine-supported principal type.
principal – Identity string, domain/name pair or wildcard value.

Returns:

Canonical principal_key and detached statements with source_kind, checkable and selection_reason fields. Invalid prospective drafts are retained separately with scope explicitly unverified.

Return type:

dict[str, Any]

Raises:

ValueError – Conflicting records share an ID, making inclusion ambiguous.

Notes

Uses engine matching and corpus admission; does not evaluate conditions or record history. Legacy principal lookup enriches detached records. Callers must serialize preparation with snapshot and prospective mutation so all inputs represent the same snapshot. No source records or cache files are rewritten.

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

Discover condition inputs and diagnostics on prepared statement copies.

Parameters:: statements – Eligible candidate records to inspect without evaluation.
Returns:: Sorted variables and per-statement clauses, variables, parse status and warnings. Unsupported clauses retain their source text and diagnostics instead of silently becoming unconditional grants.
Return type:: dict[str, Any]

Notes

Reads conditions without changing source records or engine history. Consumers may project these schemas onto an explicit checked-ID subset.

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]

Snapshot-scoped eligibility and persistence for tenancy policy limits.

class oci_policy_analysis.application.services.policy_limits_service.PolicyLimitsService(context)[source]

Bases: object

Share policy-limit rules between inventory and recommendation controls.

Parameters:: context – Current application context with repository and cache store.
Variables:: context – Snapshot whose source controls live-fetch eligibility.

property is_live_loaded: bool: Whether policy data came from a live load in the current session.

property can_fetch_live: bool: Whether this live snapshot also has an initialized OCI Limits client.

apply_manual(policies_count: str, chain_count: str) → dict[source]

Validate manual limits and persist only an active named cache.

Parameters:

policies_count – Positive whole-number policy object limit.
chain_count – Positive whole-number compartment hierarchy statement limit.

Returns:

Snapshot metadata marked as user supplied.

Return type:

dict

Raises:

ValueError – For a live-loaded snapshot or invalid values.
OSError – If the active cache cannot be updated. Both in-memory limit representations are restored on failure. Export/CIS input files and global preferences are never modified.

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 presentation adapters.

Parameters:: context – Shared application context.

STATEMENT_LIMIT = 500

NEARING_THRESHOLD = 0.85

get_dashboard_payload(*, fetch_limits: bool = True) → dict[str, Any][source]

Return full recommendations dashboard payload.

Parameters:: fetch_limits – Allow fetching missing live tenancy limits. Native UI previews disable this to keep display refreshes offline.
Returns:: Read-only recommendations sections for display.
Return type:: dict[str, Any]

refresh_analysis(*, settings: dict | None = None, on_stage=None) → dict[str, Any][source]

Rebuild recommendations and report prerequisites for the active snapshot.

Parameters:

settings – Optional current strategy/risk preferences. Only analysis settings are applied to this context; nothing is persisted.
on_stage – Optional callback receiving stage, detail and state strings.

Returns:

Dashboard over the rebuilt intelligence and corpora.

Return type:

dict[str, Any]

Notes

Runs the shared full post-load pipeline, replacing derived engines and simulation results in place, retaining local prospective statements and their CRUD manager. The caller must serialize snapshot access and rebind consumers after success or failure. Failures can leave partially rebuilt derived state. Source cache files are never written. Available OCI clients may be used by configured intelligence checks.

supersession_details(internal_id: str) → dict[str, Any] | None[source]

Return full candidate evidence and separate qualified permission coverage.

Parameters:: internal_id – Snapshot-local candidate statement identity.
Returns:: Detached candidate, finding and coverage rows, or None when the candidate no longer belongs to the current findings.
Return type:: dict[str, Any] | None

Shared principal-focused analysis workflows for UI/Web consumers.

class oci_policy_analysis.application.services.principal_analysis_service.PrincipalPolicyResult(statements: list[RegularPolicyStatement])[source]

Bases: object

Simple result wrapper for principal-oriented policy retrieval.

statements: list[RegularPolicyStatement]

class oci_policy_analysis.application.services.principal_analysis_service.PrincipalAnalysisService(context: AppContext)[source]

Bases: object

Retrieve policy statements using structured principal selectors.

AllowedSubjectType: alias of Literal[‘group’, ‘dynamic-group’, ‘any-user’, ‘any-group’, ‘service’]

by_exact_dynamic_groups(dynamic_groups: list[DynamicGroup]) → PrincipalPolicyResult[source]

Find statements matching exact dynamic groups, including legacy cache rows.

Parameters:: dynamic_groups – Dynamic groups to match exactly.
Returns:: Matched policy statements.
Return type:: PrincipalPolicyResult

by_dynamic_groups(dynamic_groups: list[DynamicGroup]) → PrincipalPolicyResult[source]

Find statements matching dynamic group principals.

Parameters:: dynamic_groups – Dynamic groups to match as principal selectors.
Returns:: Matched policy statements.
Return type:: PrincipalPolicyResult

by_exact_groups_users(*, groups: list[Group] | None = None, users: list[User] | None = None) → PrincipalPolicyResult[source]

Find statements matching exact groups and users.

Parameters:

groups – Group objects to match exactly.
users – User objects to match exactly.

Returns:

Matched policy statements.

Return type:

PrincipalPolicyResult

filter_groups_by_user(groups: list[Group], search: list[str]) → list[Group][source]

Keep groups with a known member matching a username or display name.

Parameters:

groups – Already filtered group inventory from the current snapshot.
search – Case-insensitive alternatives accepted by the user repository search.

Returns:

Matching source groups. Missing membership evidence does not establish membership; no live lookup or stored-data change occurs.

Return type:

list[Group]

user_group_rows(users: list[User]) → list[dict][source]

Describe each selected user’s memberships from the current snapshot.

Parameters:: users – Loaded user records, including their retained group IDs.
Returns:: Detached rows with user identity, group evidence, and a membership status. Unresolved IDs and users with no recorded groups remain visible. No live lookup or cache write occurs.
Return type:: list[dict]

by_groups_users(*, groups: list[Group] | None = None, users: list[User] | None = None) → PrincipalPolicyResult[source]

Find statements matching group and user principals.

User selectors are expanded by repository principal equivalence, so group-based policies for loaded user memberships are included.

Parameters:

groups – Group objects to match as principal selectors.
users – User objects to match as principal selectors.

Returns:

Matched policy statements.

Return type:

PrincipalPolicyResult

by_subject_types(*, subject_types: list[AllowedSubjectType], resource_type: str = 'Any', resource_compartment_ocid: str = '') → PrincipalPolicyResult[source]

Find statements by subject type and optional resource type scope.

Parameters:

subject_types – Subject type filters.
resource_type – Optional resource type scope; “Any” disables resource type evidence filtering.
resource_compartment_ocid – Optional request.principal.compartment.id evidence filter.

Returns:

Matched policy statements.

Return type:

PrincipalPolicyResult

by_oke_workload_identity(*, workload_namespace: str = '', workload_service_account: str = '', workload_cluster_id: str = '', subject_types: list[AllowedSubjectType] | None = None) → PrincipalPolicyResult[source]: Find statements matching OKE workload identity evidence.

static merge_unique(*statement_lists: list[RegularPolicyStatement]) → list[RegularPolicyStatement][source]

Merge statement lists while removing duplicates by internal identifier.

Parameters:: *statement_lists – One or more statement lists to merge.
Returns:: Ordered unique statements.
Return type:: list[RegularPolicyStatement]

get_resource_types() → list[str][source]

Return dynamic Resource Type values derived from policy where clauses.

Resource Type values are extracted from condition clauses using request.principal.type = '<value>' expressions.

get_workload_identity_values() → dict[str, list[str]][source]: Return discovered OKE workload identity selector values from parsed conditions.

Focused query service for MCP policy and identity tools.

class oci_policy_analysis.application.services.mcp_query_service.MCPQueryService(context: AppContext)[source]

Bases: object

Query loaded policy and identity data independently of MCP transport.

Parameters:

context – Loaded runtime with the inventory required by the requested query.

Variables:

context – Shared runtime used for repository queries.
analysis – Statement-filtering facade over the same runtime.

Notes

MCP tools are protocol adapters around these operations; Python callers do not need to start an MCP server. Queries do not fetch missing OCI data or evaluate authoritative access. Policy-only loads have limited identity data.

filter_policy_statements(filters: PolicySearch) → list[dict][source]

Filter loaded policy statements.

Parameters:: filters – Policy search criteria.
Returns:: Matching policy statement dictionaries.
Return type:: list[dict]

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

Return groups for an exact user.

Parameters:: user – Exact user descriptor.
Returns:: Groups containing the user.
Return type:: list[Group]

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

Return users for an exact group.

Parameters:: group – Exact group descriptor.
Returns:: Users in the group.
Return type:: list[User]

search_users(filters: UserSearch) → list[User][source]

Search loaded users.

Parameters:: filters – User search criteria.
Returns:: Matching users.
Return type:: list[User]

search_groups(filters: GroupSearch) → list[Group][source]

Search loaded groups.

Parameters:: filters – Group search criteria.
Returns:: Matching groups.
Return type:: list[Group]

search_dynamic_groups(filters: DynamicGroupSearch) → list[dict][source]

Search loaded dynamic groups.

Parameters:: filters – Dynamic group search criteria.
Returns:: Matching dynamic groups.
Return type:: list[dict]

list_cross_tenancy_aliases() → list[DefineStatement][source]

Return loaded cross-tenancy aliases.

Returns:: Defined aliases known to the active policy data.
Return type:: list[DefineStatement]

filter_cross_tenancy_policies_by_alias(alias: str) → list[BasePolicyStatement][source]

Return cross-tenancy statements referencing an alias.

Parameters:: alias – Alias to match.
Returns:: Matching cross-tenancy policy statements.
Return type:: list[BasePolicyStatement]

Bounded graph projection of direct policy corpus grants.

class oci_policy_analysis.application.services.policy_graph_service.PolicyGraphService(corpus: PolicyCorpus | None)[source]

Bases: object

Project direct corpus contributions into source-grant search results.

Parameters:: corpus – Compiled regular policy corpus, or None for an unloaded result.
Variables:: corpus – Snapshot bound to this service instance; construct a new service after reload.

Notes

Groups permissions by source statement/principal/path while retaining source evidence and conditions. Does not expand inherited facts or evaluate conditions and denies into an authorization decision.

get_page(*, principal: str = '', path: str = '', resource: str = '', query: str = '', effect: str = '', page: int = 1, page_size: int = 12, overview: bool = False, all_results: bool = False) → dict[str, Any][source]

Search source grants and return complete facets with selected rows.

Parameters:

principal – Case-insensitive substring of canonical or declared principal keys.
path – Case-insensitive substring of the original grant path.
resource – Substring of a resource name or expanded permission name.
query – Substring of policy name, original statement text, or statement ID.
effect – Exact effect filter, normally allow or deny; empty selects both.
page – One-based page number, clamped to at least one.
page_size – Requested rows per page, clamped to the range 1 through 12.
overview – Suppress grant rows until an effect or two-character text filter is present; counts and facets still describe all matches.
all_results – Return all matching rows on page one, overriding pagination.

Returns:

Load state, match total, page metadata, grants with complete permission lists, principal/path facets across all matches, needs_filter, and corpus summary counts. An unloaded corpus returns empty rows.

Return type:

dict[str, Any]

get_statement_evidence(grants: list[dict], *, source_statements: list[dict]) → list[dict][source]

Join selected graph combinations to complete source records and facts.

Parameters:

grants – Selected rows from get_page, including rows beyond a visible graph page when the caller requests all_results.
source_statements – Current collected and optional prospective statement views. Source IDs, policy OCIDs and prospective markers identify rows.

Returns:

Detached source records with permission_facts containing all direct corpus contributions for the selected combinations. Conditions, effect, paths and provenance remain attached to each fact. Permission contributions do not establish an OCI authorization decision.

Return type:

list[dict]

Service facade for permissions report tree/detail workflows.

This module centralizes permissions-report shaping logic so web and Tk UI layers can share the same inheritance and filtering behavior.

class oci_policy_analysis.application.services.permissions_report_service.PermissionsReportService(context: AppContext)[source]

Bases: object

Query projected permission contributions without UI dependencies.

Parameters:: context – Runtime with a full-profile permissions report; prospective views also require the active simulation engine.
Variables:: context – Shared runtime read on each report operation.

Notes

Reports retain allow/deny and conditional contributions. They do not evaluate final OCI access. A minimal load alone does not populate the full report.

get_tree(*, show_inherited_principals: bool = False, principal_query: str = '', path_query: str = '', principal_type: str = '', include_prospective: bool = False) → dict[str, Any][source]

Return a tree grouped by effective path and principal.

Parameters:

show_inherited_principals – Include principals inherited from ancestor paths.
principal_query – Whitespace-separated terms required in the principal key.
path_query – Whitespace-separated terms required in the effective path.
principal_type – Optional exact principal-type filter.
include_prospective – Include the simulation engine’s hypothetical overlay.

Returns:

Sorted paths with principal rows and available principal types.

Return type:

dict[str, Any]

get_details(*, path_key: str, subject_key: str, permission_query: str = '', include_prospective: bool = False, include_inherited: bool = False) → dict[str, Any][source]

Return allow/deny contribution rows for a path/principal pair.

Parameters:

path_key – Effective compartment path selected from the report tree.
subject_key – Canonical principal key selected from the report tree.
permission_query – Optional terms required in the permission text.
include_prospective – Include the hypothetical simulation overlay.
include_inherited – Include contributions from ancestor grant paths.

Returns:

Path, subject identity, and allow_rows/deny_rows with source evidence and condition details where available.

Return type:

dict[str, Any]

get_export_payload(*, include_prospective: bool = False) → dict[str, Any][source]

Return an export container for the selected report.

Parameters:: include_prospective – Include the hypothetical simulation overlay.
Returns:: Report tree, grant rows, exclusions, and summary counts.
Return type:: dict[str, Any]

get_export_rows() → list[dict[str, Any]][source]: Return flat report rows for CSV and tabular export workflows.

get_exclusions(*, include_prospective: bool = False) → list[dict[str, Any]][source]

Return source statements omitted from regular corpus permission facts.

Parameters:: include_prospective – Include exclusions from hypothetical statements.
Returns:: Exclusion records with source evidence and reasons.
Return type:: list[dict[str, Any]]

Shared cross-tenancy projection adapter for desktop and web consumers.

class oci_policy_analysis.application.services.cross_tenancy_analysis_service.CrossTenancyAnalysisService(repo: Any, findings: list[dict] | None = None, corpus: CrossTenancyCorpus | None = None)[source]

Bases: object

Project cross-tenancy sources and diagnostics for any consumer.

Parameters:

repo – Repository used when a precompiled corpus is unavailable.
findings – Optional intelligence findings for a fallback build.
corpus – Precompiled cross-tenancy snapshot to prefer over a fallback build.

Variables:

repo – Repository used for legacy/fallback projection.
findings – Findings associated with the fallback repository.
corpus – Optional snapshot bound to this instance.

Notes

Local relationships do not establish that matching remote policies exist. Construct a new service when the active snapshot changes.

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

Return a detached cross-tenancy projection with diagnostics.

Returns:: UI-neutral source/relationship projection and schema version. Builds a temporary corpus from the repository if no corpus was supplied.
Return type:: dict[str, Any]

static generate_single_policy(payload: dict[str, Any]) → dict[str, Any][source]

Build an ordered, copy-ready policy from valid and resolved source statements.

Parameters:: payload – Detached projection returned by get_analysis_payload().
Returns:: Policy text and counts for included, invalid, unresolved, and unclassified source statements. Source policy names and statement kinds are retained as OCI policy-language trailing comments.
Return type:: dict[str, Any]

Service facade for consolidation workbench workflows.

This module centralizes consolidation workbench orchestration so web and Tkinter consumers can gradually converge on shared service logic.

class oci_policy_analysis.application.services.consolidation_workbench_service.ConsolidationWorkbenchService(context: AppContext)[source]

Bases: object

Provide consolidation workbench operations independent from UI frameworks.

Parameters:: context – Shared application context containing repositories, cache, and engines.

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

Return consolidation status and capability flags for the current dataset.

Returns:: Status payload with tenancy info, source flags, and strategies.
Return type:: dict[str, Any]

get_strategy_catalog() → list[dict[str, str]][source]

Describe available strategies without exposing engine internals.

Returns:

Available display names and planning descriptions.: Capability restrictions match the active dataset’s strategy list.

Return type:

list[dict[str, str]]

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

Build protection browser rows from regular statements.

Returns:

One row per selectable internal ID, with source: occurrence counts. Repeated IDs are one protection unit. Source evidence is retained unchanged, including supported older caches.

Return type:

list[dict[str, Any]]

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

Load the protected statement set for the active tenancy.

Returns:: Protected set payload, or empty dict when tenancy is unavailable.
Return type:: dict[str, Any]

set_protected_set(internal_ids: list[str]) → dict[str, Any][source]

Persist protected statement IDs as a canonical protected set.

Parameters:: internal_ids – Selected statement internal IDs to mark as protected.
Returns:: Persisted protected-set payload.
Return type:: dict[str, Any]
Raises:: ValueError – If no tenancy is currently loaded.

get_candidate_rows(search: str = '') → dict[str, Any][source]

Return candidate rows and exclusion counts.

Parameters:

search – Optional free-text filter against policy name and statement text.

Returns:

Unique-ID candidate rows with source occurrence counts: and protected/invalid/system exclusions. Conflicting IDs are excluded and reported in an optional ambiguous count.

Return type:

dict[str, Any]

create_proposal(candidate_internal_ids: list[str], strategy_display_name: str) → dict[str, Any][source]

Generate and persist a consolidation proposal.

Parameters:

candidate_internal_ids – Candidate statement internal IDs selected for consolidation.
strategy_display_name – Strategy display name selected by the user.

Returns:

Generated plan and UI-oriented proposal rows.

Return type:

dict[str, Any]

Raises:

ValueError – If a selected ID represents different source statements, the strategy is unavailable, or its plan exceeds policy capacity.

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

Return run history rows with validity metadata.

Returns:: Sorted history rows with execution and conflict summary.
Return type:: list[dict[str, Any]]

get_history_run(effort_id: str) → dict[str, Any] | None[source]

Return a specific run record by effort ID.

Parameters:: effort_id – Consolidation effort identifier.
Returns:: Run record when found, otherwise None.
Return type:: dict[str, Any] | None

get_history_run_detail(effort_id: str) → dict[str, Any] | None[source]: Return one run with explicit conflict-analysis metadata.

get_conflict_analysis(plan: ConsolidationPlan | None, status: str = 'in_progress') → dict[str, Any][source]: Return the canonical conflict status for a plan without UI-specific handling.

delete_history_run(effort_id: str) → bool[source]: Delete one history run for the active tenancy.

save_plan_notes(effort_id: str, notes: str) → bool[source]

Save plan notes for a run record.

Parameters:

effort_id – Consolidation effort identifier.
notes – Notes text to persist into the plan object.

Returns:

True when update succeeded; otherwise False.

Return type:

bool

render_script(effort_id: str, fmt: str, section: str) → str[source]

Render execution/rollback script text for a run.

Parameters:

effort_id – Consolidation effort identifier.
fmt – Output format (cli or ui).
section – Requested section (execution, rollback, or both).

Returns:

Rendered script text.

Return type:

str

render_plan(plan: ConsolidationPlan, fmt: str, section: str, effort_id: str = '') → str[source]: Render a plan for either client, including a summary when it is persisted.

evaluate_progress(plan: ConsolidationPlan) → dict[str, Any][source]: Evaluate a plan against the currently loaded repository state.

check_progress(effort_id: str) → dict[str, Any][source]

Check plan progress using current repository state and marker tags.

Parameters:: effort_id – Consolidation effort identifier.
Returns:: Progress map and execution counts.
Return type:: dict[str, Any]
Raises:: ValueError – If plan is missing or dataset source is not live tenancy.

save_progress(effort_id: str, progress: dict[str, Any]) → dict[str, Any][source]

Persist plan progress and return refreshed proposal rows.

This is used when a presentation layer has already performed a live reload and only needs the common persistence/rendering behavior.

check_progress_with_refresh(effort_id: str) → dict[str, Any][source]

Check progress and refresh derived consumers after a live policy reload.

Parameters:

effort_id – Persisted plan identifier for the active tenancy.

Returns:

The same progress result as check_progress().: Cached/compliance datasets return stored progress without enrichment.

Return type:

dict[str, Any]

Notes

Callers must serialize access to the context and rebind consumers afterward, including on failure. Live reload can partially replace policy data even when it fails. Rebuilds full analysis if a permissions report was present, otherwise query foundations. Preserves prospective statements and uses existing settings. No source caches are rewritten.

Raises:

ValueError – If the plan is missing or live policy reload fails.
Exception – If rebuilding derived analysis fails after the progress check.

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

Reset consolidation state (protection + history) for active tenancy.

Returns:: Summary including cleared history count.
Return type:: dict[str, Any]
Raises:: ValueError – If no tenancy is currently loaded.

get_proposal_rows(plan: ConsolidationPlan | None, progress: dict[str, Any] | None = None, results_fallback: list[dict[str, Any]] | None = None) → list[dict[str, Any]][source]

Build UI-oriented proposal rows from plan steps.

Parameters:

plan – Consolidation plan payload.
progress – Optional progress map keyed by step ID.

Returns:

Proposal rows for workbench display.

Return type:

list[dict[str, Any]]

Service facade for policy intelligence operations.

class oci_policy_analysis.application.services.intelligence_service.IntelligenceResult(overlay: dict[str, Any])[source]

Bases: object

Intelligence result containing the engine’s current overlay.

Variables:: overlay (dict[str, Any]) – Findings and recommendations from the current engine run.

overlay: dict[str, Any]

class oci_policy_analysis.application.services.intelligence_service.IntelligenceService(context: AppContext)[source]

Bases: object

Run or inspect intelligence strategies over the loaded repository.

Parameters:

context – Runtime containing the policy repository and intelligence engine.

Variables:

context – Shared runtime read for each operation.
logger – Component logger for strategy execution.

Notes

run_all updates the intelligence engine and its permissions report. It does not replace context.corpus or context.simulation; use the post-load pipeline to rebuild all derived state together after changing repository data.

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

Run configured intelligence strategies and refresh overlay data.

Parameters:

enabled_strategy_ids – Optional subset of strategy identifiers.
params – Optional strategy parameter map.

Returns:

Latest overlay data after processing.

Return type:

IntelligenceResult

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

Return currently cached intelligence overlay data.

Returns:: Overlay payload from the intelligence engine.
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.

load_cleanup_progress(tenancy_ocid: str) → list[dict][source]

save_cleanup_progress(tenancy_ocid: str, actions: list[dict]) → None[source]: Atomically persist only this tenancy’s tracked actions and verification history.

read_consolidation_state(tenancy_ocid: str) → dict[source]

Read saved consolidation state without creating or repairing a file.

Parameters:

tenancy_ocid – Saved tenancy identifier.

Returns:

Existing state, with absent optional sections left absent.

Return type:

dict

Raises:

OSError – If the state file cannot be read.
ValueError – If JSON or section types are invalid.

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

update_tenancy_policy_limits(named_cache: str, limits: dict[str, Any]) → bool[source]: Persist operator-supplied limits with one cache, never globally.

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

Delete an unlocked cache and its index entry, preserving data on failure.

Parameters:: named_cache – Existing tenancy_label cache name; path components are rejected.
Returns:: True after removal. Locked, missing or failed operations return False.
Return type:: bool

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

Rename an indexed cache without overwriting another snapshot.

Parameters:

old_named_cache – Existing tenancy_label cache name.
new_named_cache – New tenancy_label name with no path or Windows-invalid characters.

Returns:

True after rename. Invalid names, collisions and failures return False.

Return type:

bool

Notes

Source bytes and lock state are retained. The index is replaced atomically; an exclusive hard link prevents target clobbering before the old name is removed.

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

Lock or unlock an existing cache against cleanup and manual deletion.

Parameters:

named_cache – Existing tenancy_label cache name.
preserve – True to lock, False to unlock.

Returns:

True after the index is updated; False for missing or failed operations.

Return type:

bool

update_policy_section(policy_analysis: PolicyAnalysisRepository, policy_data_reloaded: str | None, *, base_cache_name: str | None = None) → str | None[source]

Create a policy-only reload snapshot from a full cache without changing that original snapshot. The new name is <base>_reload-01 (then -02, and so on). It contains fresh policy/compartment data and retained IAM data, and is marked so historical comparison can avoid presenting IAM diffs as fresh data.

Parameters:

policy_analysis – PolicyAnalysisRepository with fresh policy/compartment data in memory
policy_data_reloaded – ISO timestamp string for reloaded policy data.
base_cache_name – Cache that supplied the IAM portion, when the caller knows it. Older callers may omit this value.

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, *, verbose_override: bool | None = None) → 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.
verbose_override – Explicit runtime verbosity lock. None retains legacy root-DEBUG inference; False permits restoring saved DEBUG preferences.

oci_policy_analysis.application.core.support.logger.set_component_level(component: str, level: str | int, *, verbose_override: bool | None = None) → 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.
verbose_override – Explicit runtime lock; None preserves root-DEBUG inference.

Snapshot-bound prospective orchestration shared by desktop evidence views.

class oci_policy_analysis.application.services.prospective_workbench_service.ProspectiveWorkbenchService(context)[source]

Bases: object

Bind persisted what-if records to the active context without changing sources.

Parameters:

context – Loaded AppContext; create a new facade after a snapshot load.

Variables:

context – Current loaded runtime.
records – Tenancy-scoped CRUD and persistence service shared with other adapters.

Notes

Construction loads prospects once and hydrates simulation without writing caches. Risk calculations use a detached statement and repository view.

statement_views() → list[dict][source]

Return detached, labelled prospects, including invalid source evidence.

Returns:: Statement-shaped records suitable for core search filters.
Return type:: list[dict]

potential_risk(record) → dict | None[source]

Estimate risk with existing scoring over detached validated evidence.

Parameters:

record – Validated ProspectiveStatementRecord from this tenancy.

Returns:

Score, notes and recommendations, or None if invalid.: This heuristic does not evaluate conditions or establish OCI access.

Return type:

dict | None

Application service facade for condition tester workflows.

This service exposes a stable API for UI/API consumers while delegating core parsing and evaluation behavior to oci_policy_analysis.application.core.parser.

oci_policy_analysis.application.services.condition_tester_service.extract_condition_clause(condition: str = '', statement_text: str = '') → str[source]

Extract a tester expression from a clause or the original policy statement.

Quoted names and string values containing where are handled by the policy lexer. A leading where keyword is omitted from standalone clauses.

class oci_policy_analysis.application.services.condition_tester_service.ConditionEvaluationResult(condition_string: str, policy_result: str, log: list[dict[str, Any]], granted: bool)[source]

Bases: object

Structured result payload for condition evaluation responses.

Variables:

condition_string (str) – The evaluated condition clause.
policy_result (str) – GRANTED or DENIED.
log (list[dict[str, Any]]) – Structured comparison/evaluation log entries.
granted (bool) – Boolean result equivalent of policy_result.

condition_string: str

policy_result: str

log: list[dict[str, Any]]

granted: bool

class oci_policy_analysis.application.services.condition_tester_service.ConditionTesterService[source]

Bases: object

Service that provides format/extract/evaluate operations for conditions.

format_clause(clause: str) → str[source]

Format a condition clause for readability.

Parameters:: clause – Raw clause text.
Returns:: Beautified clause text.

extract_variables(clause: str) → list[str][source]

Extract variable names from a condition clause.

Parameters:: clause – Condition clause text.
Returns:: Sorted list of discovered variable names.

parse_clause(clause: str) → dict[str, Any][source]

Parse entered conditions or a full policy and discover both operand inputs.

Parameters:: clause – Condition expression, optional leading where, or full policy.
Returns:: Extracted clause, display structure including syntax diagnostics, and sorted variable names. Invalid/absent expressions have no input fields and must not be treated as evaluated denials.
Return type:: dict[str, Any]

evaluate(clause: str, variables: dict[str, Any]) → ConditionEvaluationResult[source]

Evaluate a condition clause against simulated variables.

Parameters:

clause – Condition clause text.
variables – Simulated variable map.

Returns:

Structured evaluation result payload.

Notes

The returned policy_result is normalized to GRANTED/DENIED even when lower-level parser output omits that field.

Explain simulation runs and prepare reviewable what-if policy candidates.

class oci_policy_analysis.application.services.simulation_workbench_service.SimulationWorkbenchService(context)[source]

Bases: object

Build snapshot-bound explanations, related requests and grant candidates.

Parameters:: context – Loaded AppContext, serialized by the caller during execution.

Notes

Execution records engine history. Other methods are read-only. Candidates are validated local drafts, not OCI changes or guarantees of access. Reverse mappings use bundled reference data; corpus facts supply retained provenance, not an authoritative decision or a complete OCI permission map.

execute(request: dict) → dict[source]

Validate explicit inclusion, execute and retain the original run inputs.

Parameters:: request – Environment, qualified operation, explicit checked IDs and condition values from the request screen. An empty ID list stays empty.
Returns:: Immutable-by-convention request/result evidence with permission rows, condition comparisons, findings and related scenario choices.
Return type:: dict
Raises:: ValueError – Unknown operation, stale environment or ineligible IDs.

history() → list[dict][source]

Return detached history for the current snapshot, including legacy runs.

Returns:: Original engine runs with optional workbench_report evidence. Snapshot replacement creates a fresh engine and history; no cache is written.
Return type:: list[dict]

related_scenarios(operation: str) → list[dict][source]

Resolve advisory dependencies into valid catalog operation choices.

Parameters:: operation – Qualified or unambiguous reference operation name.
Returns:: Canonical operation labels, applicability and dependency permissions. The caller must select a target path and reload statements. Unknown/ambiguous operation names are not presented as executable choices.
Return type:: list[dict]

suggestion_metadata() → dict[source]

Return supported grant principals and paths for explicit user choices.

Returns:: Shared builder choices filtered to OCI grant principal types. User simulations require choosing a grant-bearing group or other principal.
Return type:: dict

policy_candidates(report: dict, *, principal_key: str, granularity: str, effective_path: str, policy_path: str = 'ROOT', where_clause: str = '') → dict[source]

Reverse missing permissions into validated, explicitly scoped policy drafts.

Parameters:

report – Prior run evidence; only its missing primary permissions are used.
principal_key – Explicit grant principal from suggestion_metadata.
granularity – permission, resource, or family. Resource/family drafts may grant additional catalog permissions, which are listed separately.
effective_path – User-selected target path, including descendant inheritance.
policy_path – Ancestor compartment where the hypothetical policy is stored.
where_clause – Optional custom condition expression, without where keyword.

Returns:

Validated candidates, unmapped permissions, corpus evidence and limitations. Multiple statements may be needed; each candidate carries its covered and additional permissions and source mapping.

Return type:

dict

Raises:

ValueError – Unsupported principal, scope, condition or granularity.

Notes

Uses greedy catalog coverage, not a proof of globally minimal policy. No generated policy is saved or applied. Matching denies still win.

On-demand report generation services.

class oci_policy_analysis.application.services.reports_service.ReportsService(context: AppContext)[source]

Bases: object

Generate explicit, read-only reports without changing normal analysis.

Parameters:: context – Shared application context.

static catalog() → list[dict[str, str]][source]

Return the report choices and coverage guidance shared by adapters.

Returns:: Fresh report IDs, labels and prerequisite notes.
Return type:: list[dict[str, str]]

export_formats(report: dict[str, Any]) → dict[str, str][source]

Serialize one generated report as JSON, Markdown and portable HTML.

Parameters:: report – Native, snapshot-bound report evidence; never modified.
Returns:: UTF-8-ready json, markdown and html document strings.
Return type:: dict[str, str]

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

Run the expensive overlap analysis and return its full JSON report.

Returns:: Metadata and all candidate-to-overlap findings.
Return type:: dict[str, Any]

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

Return the loaded compartment, policy, and statement hierarchy.

Returns:: Inventory grouped as compartment, policy, and statements.
Return type:: dict[str, Any]

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

Return the existing effective-permissions report as an export artifact.

Equivalent grants are consolidated for this report only. The native permissions report deliberately retains every source row for detail and CSV workflows; an on-demand report is more useful when it presents one effective grant with its contributing statements collected beneath it.

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

Return complete-supersession findings grouped by effective path.

Supersession analysis runs with the normal policy-intelligence strategies. This report is a read-only rendering of that evidence; it does not rerun or alter analysis when the user opens it.

get_report(report_id: str) → dict[str, Any][source]: Generate one supported report by its stable identifier.

to_markdown(report: dict[str, Any]) → str[source]

Convert a report’s native JSON structure into readable Markdown.

Parameters:: report – Generated report payload.
Returns:: Deterministic Markdown suitable for preview or download.
Return type:: str

with_rendered_formats(report: dict[str, Any]) → dict[str, Any][source]

Attach Markdown and safe preview HTML to a native report payload.

Parameters:: report – Native JSON report payload.
Returns:: Report payload with Markdown and rendered HTML.
Return type:: dict[str, Any]

Historical cache comparison service with stable-key normalization.

class oci_policy_analysis.application.services.historical_analysis_service.HistoricalDiffItem(action: str, section: str, stable_key: str, title: str, old_value: dict[str, Any] | None = None, new_value: dict[str, Any] | None = None, changed_fields: list[dict[str, Any]] | None = None)[source]

Bases: object

Describe one added, removed, or modified item in a cache comparison.

action: str

section: str

stable_key: str

title: str

old_value: dict[str, Any] | None = None

new_value: dict[str, Any] | None = None

changed_fields: list[dict[str, Any]] | None = None

class oci_policy_analysis.application.services.historical_analysis_service.HistoricalSectionResult(section: str, key: str, added: int, removed: int, modified: int, items: list[HistoricalDiffItem])[source]

Bases: object

Summarize changes for one historical comparison section.

section: str

key: str

added: int

removed: int

modified: int

items: list[HistoricalDiffItem]

class oci_policy_analysis.application.services.historical_analysis_service.HistoricalCompareResult(left_cache: str, right_cache: str, policy_sections: list[~oci_policy_analysis.application.services.historical_analysis_service.HistoricalSectionResult], identity_sections: list[~oci_policy_analysis.application.services.historical_analysis_service.HistoricalSectionResult], identity_comparable: bool = True, skipped_sections: dict[str, str] = <factory>, metadata: dict[str, ~typing.Any] = <factory>, section_warnings: dict[str, str] = <factory>)[source]

Bases: object

Contain the normalized results of comparing two cache snapshots.

left_cache: str

right_cache: str

policy_sections: list[HistoricalSectionResult]

identity_sections: list[HistoricalSectionResult]

identity_comparable: bool = True

skipped_sections: dict[str, str]

metadata: dict[str, Any]

section_warnings: dict[str, str]

class oci_policy_analysis.application.services.historical_analysis_service.HistoricalAnalysisService(cache_manager)[source]

Bases: object

Compare two cache snapshots and return normalized sectioned results.

POLICY_SECTIONS: list[tuple[str, str]] = [('Policies', 'policies'), ('Regular Statements', 'policy_statements'), ('Defined Aliases', 'defined_aliases'), ('Cross-Tenancy Statements', 'cross_tenancy_statements')]

IDENTITY_SECTIONS: list[tuple[str, str]] = [('Compartments', 'compartments'), ('Identity Domains', 'identity_domains'), ('Groups', 'groups'), ('Dynamic Groups', 'dynamic_groups'), ('Users', 'users')]

static comparison_value(value: Any, *, meaningful_only: bool = False) → Any[source]

Return detached comparable JSON without ephemeral implementation fields.

Parameters:

value – JSON source value; never changed in place.
meaningful_only – Also omit entity-level internal_id and stable_key fields. OCI IDs, timestamps, conditions, validity and array order remain.

Returns:

Detached comparison value. Underscore-prefixed runtime fields are omitted only at the entity level. Nested JSON, including arbitrary OCI tag keys, is preserved literally.

Return type:

Any

static snapshot_kind(snapshot: dict[str, Any]) → str[source]

classmethod describe_snapshot(snapshot: dict[str, Any]) → str[source]

compare_caches(*, left_cache: str, right_cache: str, meaningful_only: bool = False, include_incomplete: bool = False) → HistoricalCompareResult[source]

Read two caches and compare their sections without changing active data.

Parameters:

left_cache – Baseline cache name.
right_cache – Target cache name.
meaningful_only – Ignore ephemeral internal_id and stable_key fields.
include_incomplete – Compare observed data despite incomplete-collection metadata, with warnings; differences do not establish OCI changes.

Returns:

Sectioned changes and coverage limitations.

Return type:

HistoricalCompareResult

compare_snapshots(left: dict, right: dict, *, left_name: str = 'Baseline', right_name: str = 'Target', meaningful_only: bool = False, include_incomplete: bool = False) → HistoricalCompareResult[source]

Compare detached JSON snapshots using the same cache compatibility rules.

Parameters:

left – Baseline combined-cache JSON.
right – Target combined-cache JSON.
left_name – Display name for the baseline.
right_name – Display name for the target.
meaningful_only – Omit ephemeral internal IDs; preserve original evidence.
include_incomplete – Compare available sections even when collection was incomplete. Missing target records are unverified snapshot differences.

Returns:

Changes plus per-section non-comparison reasons.

Return type:

HistoricalCompareResult

Raises:

ValueError – Either snapshot is missing or known tenancy OCIDs differ.

Notes

List items align by stable keys, with legacy fallback identities. An identity change can appear as removal/addition rather than modification. Meaningful mode first aligns exact statement text within a policy, then matches comment-only edits as modifications, retaining duplicate occurrences and original evidence. Grant/condition edits appear as removal/addition. Nested JSON arrays retain their order. No source files or engines change.

Service facade for cache persistence operations.

class oci_policy_analysis.application.services.cache_service.CacheService(cache: CacheManager)[source]

Bases: object

Expose cache operations to UI/API without direct CacheManager usage.

cache: CacheManager

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

List available cache names.

Parameters:: tenancy_name – Optional tenancy name filter.
Returns:: Cache names matching the filter.
Return type:: list[str]

get_preserved_cache_set() → set[str][source]

Fetch the set of preserved cache names.

Returns:: Preserved cache names.
Return type:: set[str]

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

List cache names and their current lock state without reading snapshot data.

Returns:: Names and preserved/locked flags from the cache index.
Return type:: list[dict[str, Any]]

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

List saved consolidation tenancies and counts without loading inventory.

Returns:: Tenancy IDs with plan and protection counts.
Return type:: list[dict[str, Any]]

reset_consolidation(tenancy_ocid: str) → dict[str, Any][source]

Clear one tenancy’s local protection and complete consolidation history.

Parameters:: tenancy_ocid – Explicitly selected tenancy; must be nonempty.
Returns:: Tenancy ID and number of removed history records.
Return type:: dict[str, Any]
Raises:: ValueError – If no tenancy is supplied.

Notes

Uses the existing canonical state format. Inventory caches and OCI policies are unchanged. Removes saved notes, progress, and rollback snapshots along with their plan records; no migration is required.

set_locked(cache_name: str, locked: bool) → bool[source]

Preserve or release a saved cache for automatic and explicit cleanup.

Parameters:

cache_name – Existing saved cache name.
locked – True to protect the cache, False to allow deletion.

Returns:

Whether the cache lock was updated successfully.

Return type:

bool

load_cache(policy_repo, cache_name: str) → str[source]

Load a named cache into the provided policy repository.

Parameters:

policy_repo – Policy repository receiving loaded data.
cache_name – Cache entry name.

Returns:

Underlying cache manager status/result.

Return type:

str

load_cache_json(cache_name: str) → dict[str, Any][source]

Load a named cache file as JSON.

Parameters:: cache_name – Cache entry name.
Returns:: Loaded JSON content.
Return type:: dict[str, Any]

remove_cache(cache_name: str) → bool[source]

Remove a cache entry.

Parameters:: cache_name – Cache entry name.
Returns:: True when a cache entry was removed.
Return type:: bool

rename_cache(old_cache_name: str, new_cache_name: str) → bool[source]

Rename an existing cache entry.

Parameters:

old_cache_name – Existing cache name.
new_cache_name – New cache name.

Returns:

True when rename succeeded.

Return type:

bool

save_cache(policy_repo, export_file: Any | None = None, preserved: bool = False) → str[source]

Save current repository state into cache storage.

Parameters:

policy_repo – Policy repository source.
export_file – Optional explicit export destination.
preserved – Whether to mark cache as preserved.

Returns:

Saved cache entry name or status.

Return type:

str

import_from_json(*, loaded_json: dict[str, Any], policy_repo: Any) → bool[source]

Import repository state from pre-loaded JSON payload.

Parameters:

loaded_json – Parsed JSON payload.
policy_repo – Policy repository target.

Returns:

True when import succeeds.

Return type:

bool

update_policy_section(policy_repo: Any, *, policy_data_reloaded: str | None = None, base_cache_name: str | None = None) → str | None[source]

Create a policy-reload cache from an existing cache snapshot.

Parameters:

policy_repo – Policy repository source.
policy_data_reloaded – Optional policy data indicator.
base_cache_name – Cache that supplied the retained IAM data.

Returns:

New cache path, or None when no cache could be created.

get_consolidation_state(tenancy_ocid: str) → dict[source]

Get or create consolidation state for a tenancy.

Parameters:: tenancy_ocid – Tenancy OCID key.
Returns:: Consolidation state object.
Return type:: dict

save_consolidation_state(tenancy_ocid: str, state: dict) → None[source]

Persist consolidation state for a tenancy.

Parameters:

tenancy_ocid – Tenancy OCID key.
state – Consolidation state payload.

Returns:

None

Desktop usage sessions with non-blocking, best-effort publication.

class oci_policy_analysis.application.services.usage_service.UsageSession(settings: dict, app_version: str)[source]

Bases: object

Own Qt tracking lifecycle while preserving legacy run/event JSON fields.

Parameters:

settings – Shared preferences; an absent enabled flag defaults to True.
app_version – Installed application version for the run document.

Notes

Only fixed UI identifiers and numeric counts are accepted here. Uploads run off the UI thread. Opt-out discards pending data; in-flight HTTP requests cannot be recalled. No disk queue is retained.

set_enabled(enabled: bool) → None[source]: Apply opt-out immediately, or begin a fresh run on opt-in.

set_tenancy(tenancy_ocid: str | None) → None[source]: Group runs by the existing six-character suffix, never a full OCID.

view(name: str, sub_view: str | None = None) → None[source]: Record code-defined page identifiers; callers must not pass user text.

operation(name: str, *, source: str = '', success: bool = True, counts: dict | None = None) → None[source]: Record a fixed operation category, source enum and inventory counts.

close(*, wait: bool = True) → None[source]: Queue one final upload; optionally wait outside the GUI event loop.

Bounded, process-local OCI call metadata independent of logging and snapshots.

class oci_policy_analysis.application.core.support.api_calls.ApiCallJournal(capacity=5000)[source]

Bases: object

Retain bounded metadata from concurrent API calls without writing to disk.

Parameters:: capacity – Maximum completed call records retained in memory.

Notes

Contains metadata only: no request/response bodies, credentials, or raw headers. Sequence numbers survive clearing so GUI cursors remain valid. Pagination/retry attempts are separate records when using OCI SDK clients.

append(record)[source]

Append a completed call, evicting older successes before errors when full.

Parameters:: record – Scalar diagnostic metadata produced by the call observer.

snapshot(revision=None)[source]: Return a revision, detached rows, and eviction count under one lock.

clear()[source]: Forget process-local metadata without touching application log files.

oci_policy_analysis.application.core.support.api_calls.api_error_summary(error: Exception) → str[source]

Return a concise error reason without SDK troubleshooting boilerplate.

Parameters:: error – OCI ServiceError or a local collection exception.
Returns:: HTTP status, service code, reason, and request ID where available.
Return type:: str

class oci_policy_analysis.application.core.support.api_calls.ApiCallObservation(label, fn, args, kwargs, tenancy='')[source]

Bases: object

Observe a repository operation and each underlying SDK HTTP attempt.

Parameters:

label – Application operation label.
fn – Bound SDK method or pagination helper.
args – Positional call arguments; used only to find a paginated SDK method.
kwargs – Call arguments; only allowlisted resource IDs enter metadata.
tenancy – Tenancy ID for associating calls across snapshot reloads.

Notes

Finish in a finally block. Test doubles/non-SDK helpers produce one logical-operation record; no invented HTTP status is assigned to them.

finish(response=None, error=None)[source]: Record a fallback operation when no SDK attempts were observed, then detach.

Shared tag-based policy query helpers.

class oci_policy_analysis.application.services.tag_based_policy_service.TagBasedPolicyService[source]

Bases: object

Centralized parsed-condition support for tag-based policy queries.

SEMANTICS_BY_ACCESS_TYPE = {'request.principal.compartment': 'requestor_compartment_tag', 'request.principal.group': 'requestor_group_tag', 'target.resource': 'target_resource_tag', 'target.resource.compartment': 'target_compartment_tag'}

TARGET_RESOURCE_WARNINGS = ('target.resource.tag conditions may require a separate list or inspect policy to discover tagged resources.', 'target.resource.tag conditions cannot grant create permissions because the target resource tag does not exist yet.', 'target.resource.tag support has service-specific limitations; verify the target service documents tag conditions.')

TARGET_COMPARTMENT_WARNINGS = ('target.resource.compartment.tag applies through nested compartments, so child compartment resources can match parent compartment tags.',)

classmethod enrich_statement(statement: RegularPolicyStatement) → RegularPolicyStatement[source]: Attach normalized tag-condition and warning fields to a statement.

classmethod enrich_statements(statements: list[RegularPolicyStatement]) → list[RegularPolicyStatement][source]

classmethod matches_tag_filters(statement: RegularPolicyStatement, filters: PolicySearch) → bool[source]: Return true when one parsed tag condition satisfies all supplied tag filters.

classmethod matches_condition_atom_terms(statement: RegularPolicyStatement, terms_value: object) → bool[source]

classmethod matches_policy_metadata_tags(statement: RegularPolicyStatement, field: str, terms_value: object) → bool[source]

classmethod query(statements: list[RegularPolicyStatement], filters: PolicySearch | None = None) → dict[str, Any][source]: Return filtered tag-oriented statements with summary metadata.

classmethod inventory(statements: list[RegularPolicyStatement]) → dict[str, Any][source]

Project read-only tag evidence, including unparsed source references.

Parameters:: statements – Current regular statement records. Originals are never enriched or rewritten as a side effect of this query.
Returns:: Detached tag statements, one row per parsed predicate (or an explicit unparsed reference), and unique namespace counts. Predicates retain the full condition and original statement evidence; an atom alone is not proof that a complete policy condition passes.
Return type:: dict[str, Any]

classmethod context_warnings_for_conditions(tag_conditions: list[dict[str, Any]]) → list[str][source]

Reference discovery contracts

Strict, structured contracts for bundled reference catalog lookups.

class oci_policy_analysis.application.core.models.models_reference_lookup.ReferenceQuery(*, query: Annotated[str, MinLen(min_length=1), MaxLen(max_length=200)], match: Literal['search', 'exact'] = 'search', limit: Annotated[int, Ge(ge=1), Le(le=25)] = 10, cursor: Annotated[str | None, MaxLen(max_length=512)] = None)[source]

Bases: BaseModel

Validate bounded discovery inputs without allowing unrecognized fields.

model_config = {'extra': 'forbid', 'str_strip_whitespace': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

query: str

match: Literal['search', 'exact']

limit: int

cursor: str | None

class oci_policy_analysis.application.core.models.models_reference_lookup.ResourceReferenceQuery(*, query: Annotated[str, MinLen(min_length=1), MaxLen(max_length=200)], match: Literal['search', 'exact'] = 'search', limit: Annotated[int, Ge(ge=1), Le(le=25)] = 10, cursor: Annotated[str | None, MaxLen(max_length=512)] = None, kind: Literal['resource', 'family', 'either'] = 'either', verb: Literal['inspect', 'read', 'use', 'manage'] | None = None)[source]

Bases: ReferenceQuery

Find resources or families and optionally expand one cumulative IAM verb.

kind: Literal['resource', 'family', 'either']

verb: Literal['inspect', 'read', 'use', 'manage'] | None

model_config = {'extra': 'forbid', 'str_strip_whitespace': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class oci_policy_analysis.application.core.models.models_reference_lookup.OperationReferenceQuery(*, query: Annotated[str, MinLen(min_length=1), MaxLen(max_length=200)], match: Literal['search', 'exact'] = 'search', limit: Annotated[int, Ge(ge=1), Le(le=25)] = 10, cursor: Annotated[str | None, MaxLen(max_length=512)] = None, api_group: Annotated[str | None, MinLen(min_length=1), MaxLen(max_length=100)] = None)[source]

Bases: ReferenceQuery

Find qualified API operations, optionally within one catalog API group.

api_group: str | None

model_config = {'extra': 'forbid', 'str_strip_whitespace': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class oci_policy_analysis.application.core.models.models_reference_lookup.PermissionReferenceQuery(*, query: Annotated[str, MinLen(min_length=1), MaxLen(max_length=200)], match: Literal['search', 'exact'] = 'search', limit: Annotated[int, Ge(ge=1), Le(le=25)] = 10, cursor: Annotated[str | None, MaxLen(max_length=512)] = None, resource: Annotated[str | None, MinLen(min_length=1), MaxLen(max_length=200)] = None)[source]

Bases: ReferenceQuery

Find permissions and reverse mappings, optionally for one exact resource.

resource: str | None

model_config = {'extra': 'forbid', 'str_strip_whitespace': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class oci_policy_analysis.application.core.models.models_reference_lookup.ReferenceRecord(*, canonical_name: str, match_type: ~typing.Literal['identifier', 'alias', 'search_term', 'text'], matched_text: str, notes: str = '', source_urls: list[str] = <factory>, catalog_status: str = 'unreviewed', mapping_status: str = 'mapped')[source]

Bases: BaseModel

Common match identity, discovery evidence and catalog provenance.

model_config = {'extra': 'forbid'}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

canonical_name: str

match_type: Literal['identifier', 'alias', 'search_term', 'text']

matched_text: str

notes: str

source_urls: list[str]

catalog_status: str

mapping_status: str

class oci_policy_analysis.application.core.models.models_reference_lookup.PermissionExpansion(*, permissions: list[str], total_count: int, truncated: bool)[source]

Bases: BaseModel

A bounded cumulative permission list, with an explicit total and truncation.

permissions: list[str]

total_count: int

truncated: bool

model_config = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class oci_policy_analysis.application.core.models.models_reference_lookup.ResourceReferenceMatch(*, canonical_name: str, match_type: ~typing.Literal['identifier', 'alias', 'search_term', 'text'], matched_text: str, notes: str = '', source_urls: list[str] = <factory>, catalog_status: str = 'unreviewed', mapping_status: str = 'mapped', kind: ~typing.Literal['resource', 'family'], source_resource_name: str, aliases: list[str] = <factory>, search_terms: list[str] = <factory>, families: list[str] = <factory>, member_resources: list[str] = <factory>, member_count: int = 0, members_truncated: bool = False, unmapped_member_resources: list[str] = <factory>, unmapped_member_count: int = 0, supported_verbs: list[str] = <factory>, permissions_by_verb: dict[~typing.Literal['inspect', 'read', 'use', 'manage'], ~oci_policy_analysis.application.core.models.models_reference_lookup.PermissionExpansion])[source]

Bases: ReferenceRecord

Resource/family identity, membership and cumulative allow-verb expansion.

kind: Literal['resource', 'family']

source_resource_name: str

aliases: list[str]

search_terms: list[str]

families: list[str]

member_resources: list[str]

member_count: int

members_truncated: bool

unmapped_member_resources: list[str]

unmapped_member_count: int

supported_verbs: list[str]

permissions_by_verb: dict[Literal['inspect', 'read', 'use', 'manage'], PermissionExpansion]

model_config = {'extra': 'forbid'}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class oci_policy_analysis.application.core.models.models_reference_lookup.RelatedReferenceCheck(*, resource: str = '', permissions: list[str] = <factory>, operation: str = '', applies_when: str = '', reason: str = '', **extra_data: Any)[source]

Bases: BaseModel

Conditional or related API requirements retained from the catalog.

model_config = {'extra': 'allow'}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

resource: str

permissions: list[str]

operation: str

applies_when: str

reason: str

class oci_policy_analysis.application.core.models.models_reference_lookup.OperationReferenceMatch(*, canonical_name: str, match_type: ~typing.Literal['identifier', 'alias', 'search_term', 'text'], matched_text: str, notes: str = '', source_urls: list[str] = <factory>, catalog_status: str = 'unreviewed', mapping_status: str = 'mapped', api_group: str, operation_name: str, request_operation: str, permissions: list[str], permission_count: int, permissions_truncated: bool, unmapped_permissions: list[str], related_checks: list[~oci_policy_analysis.application.core.models.models_reference_lookup.RelatedReferenceCheck] = <factory>)[source]

Bases: ReferenceRecord

Qualified API identity, known requirements and unresolved resource mappings.

api_group: str

operation_name: str

request_operation: str

permissions: list[str]

permission_count: int

permissions_truncated: bool

unmapped_permissions: list[str]

related_checks: list[RelatedReferenceCheck]

model_config = {'extra': 'forbid'}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class oci_policy_analysis.application.core.models.models_reference_lookup.PermissionResourceMatch(*, resource: str, source_resource_name: str, families: list[str], lowest_granting_verb: Literal['inspect', 'read', 'use', 'manage'], granting_verbs: list[Literal['inspect', 'read', 'use', 'manage']], additional_permissions: PermissionExpansion, source_urls: list[str])[source]

Bases: BaseModel

Lowest catalog granting verb for one resource, not a complete policy plan.

resource: str

source_resource_name: str

families: list[str]

lowest_granting_verb: Literal['inspect', 'read', 'use', 'manage']

granting_verbs: list[Literal['inspect', 'read', 'use', 'manage']]

additional_permissions: PermissionExpansion

source_urls: list[str]

model_config = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class oci_policy_analysis.application.core.models.models_reference_lookup.PermissionReferenceMatch(*, canonical_name: str, match_type: ~typing.Literal['identifier', 'alias', 'search_term', 'text'], matched_text: str, notes: str = '', source_urls: list[str] = <factory>, catalog_status: str = 'unreviewed', mapping_status: str = 'mapped', resource_mappings: list[~oci_policy_analysis.application.core.models.models_reference_lookup.PermissionResourceMatch], resource_mapping_count: int, resource_mappings_truncated: bool, operations: list[str], operation_count: int, operations_truncated: bool)[source]

Bases: ReferenceRecord

Permission evidence from resource tables and operation requirements.

resource_mappings: list[PermissionResourceMatch]

resource_mapping_count: int

resource_mappings_truncated: bool

operations: list[str]

operation_count: int

operations_truncated: bool

model_config = {'extra': 'forbid'}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class oci_policy_analysis.application.core.models.models_reference_lookup.ResourceReferenceResult(*, catalog_id: str, catalog_version: str, total_matches: int, matches: list[ResourceReferenceMatch], next_cursor: str | None, warnings: list[str])[source]

Bases: BaseModel

Structured resource discovery and expansion result.

model_config = {'extra': 'forbid'}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

catalog_id: str

catalog_version: str

total_matches: int

matches: list[ResourceReferenceMatch]

next_cursor: str | None

warnings: list[str]

class oci_policy_analysis.application.core.models.models_reference_lookup.OperationReferenceResult(*, catalog_id: str, catalog_version: str, total_matches: int, matches: list[OperationReferenceMatch], next_cursor: str | None, warnings: list[str])[source]

Bases: BaseModel

Structured API operation discovery result.

model_config = {'extra': 'forbid'}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

catalog_id: str

catalog_version: str

total_matches: int

matches: list[OperationReferenceMatch]

next_cursor: str | None

warnings: list[str]

class oci_policy_analysis.application.core.models.models_reference_lookup.PermissionReferenceResult(*, catalog_id: str, catalog_version: str, total_matches: int, matches: list[PermissionReferenceMatch], next_cursor: str | None, warnings: list[str])[source]

Bases: BaseModel

Structured permission discovery and reverse mapping result.

model_config = {'extra': 'forbid'}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

catalog_id: str

catalog_version: str

total_matches: int

matches: list[PermissionReferenceMatch]

next_cursor: str | None

warnings: list[str]

Bounded, bidirectional reference discovery over an independently loaded catalog.

class oci_policy_analysis.application.services.reference_lookup_service.ReferenceLookupService(reference_data: ReferenceDataRepo)[source]

Bases: object

Discover catalog terminology and reverse mappings without loading a tenancy.

Parameters:

reference_data – Loaded catalog retained for this service’s lifetime.

Variables:

catalog_id – Content fingerprint used to bind continuation cursors.
catalog_version – Installed package version; the fingerprint identifies edits.

Notes

Treat the supplied repository as immutable. Results are detached models. Recreate the service after changing catalog data. No OCI calls, settings, snapshot loads or cache writes occur here.

resource_reference(query: ResourceReferenceQuery) → ResourceReferenceResult[source]

Find resource/family names and cumulative allow-verb permission evidence.

Parameters:: query – Validated search, kind, verb and page controls.
Returns:: Ranked identities, memberships and bounded expansions.
Return type:: ResourceReferenceResult
Raises:: ValueError – A continuation cursor does not match this request/catalog.

operation_reference(query: OperationReferenceQuery) → OperationReferenceResult[source]

Find qualified operations with fixed, conditional and unmapped requirements.

Parameters:: query – Validated wording or exact name, optional API group and page controls.
Returns:: Qualified names, source notes and bounded requirements.
Return type:: OperationReferenceResult
Raises:: ValueError – A continuation cursor does not match this request/catalog.

permission_reference(query: PermissionReferenceQuery) → PermissionReferenceResult[source]

Find named permissions and reverse resource/verb and operation mappings.

Parameters:: query – Validated permission search with an optional exact resource filter.
Returns:: Lowest granting verbs and retained mapping gaps.
Return type:: PermissionReferenceResult
Raises:: ValueError – Resource is unknown or a cursor does not match this request/catalog.