Policy Recommendations and Consolidation

Recommendation strategies

class oci_policy_analysis.application.core.engine.intelligence_strategies.risk.RiskScoreStrategy(strategy_id: str = 'risk_scores', display_name: str = 'Risk scores', category: str = 'risk')

Bases: object

Intelligence strategy: calculate potential risk scores for policy statements.

strategy_id: str = 'risk_scores'

display_name: str = 'Risk scores'

category: str = 'risk'

run(repo: PolicyAnalysisRepository, overlay: dict, params: dict | None = None) → None

class oci_policy_analysis.application.core.engine.intelligence_strategies.overlap.OverlapStrategy(strategy_id: str = 'overlap', display_name: str = 'Policy overlap', category: str = 'overlap')

Bases: object

Intelligence strategy: analyze policy statement overlaps.

strategy_id: str = 'overlap'

display_name: str = 'Policy overlap'

category: str = 'overlap'

run(repo: PolicyAnalysisRepository, overlay: dict, params: dict | None = None) → None

class oci_policy_analysis.application.core.engine.intelligence_strategies.consolidation_suggestion.ConsolidationSuggestionStrategy(strategy_id: str = 'consolidation_suggestion', display_name: str = 'Consolidation suggestions', category: str = 'consolidation_suggestion')

Bases: object

Intelligence strategy: suggest policy consolidation opportunities.

strategy_id: str = 'consolidation_suggestion'

display_name: str = 'Consolidation suggestions'

category: str = 'consolidation_suggestion'

run(repo: PolicyAnalysisRepository, overlay: dict, params: dict | None = None) → None

Cleanup strategies

class oci_policy_analysis.application.core.engine.intelligence_strategies.cleanup_anyuser_no_where.AnyuserNoWhereCheck(strategy_id: str = 'anyuser_no_where', display_name: str = 'Any-user without where', category: str = 'cleanup')

Bases: object

Intelligence strategy: collect any-user statements with no where clause.

strategy_id: str = 'anyuser_no_where'

display_name: str = 'Any-user without where'

category: str = 'cleanup'

run(repo: PolicyAnalysisRepository, overlay: dict, params: dict | None = None) → None

class oci_policy_analysis.application.core.engine.intelligence_strategies.cleanup_invalid.InvalidStatementsCheck(strategy_id: str = 'invalid_statements', display_name: str = 'Invalid statements', category: str = 'cleanup')

Bases: object

Intelligence strategy: collect invalid policy statements for cleanup.

strategy_id: str = 'invalid_statements'

display_name: str = 'Invalid statements'

category: str = 'cleanup'

run(repo: PolicyAnalysisRepository, overlay: dict, params: dict | None = None) → None

class oci_policy_analysis.application.core.engine.intelligence_strategies.cleanup_statements_too_open.StatementsTooOpenCheck(strategy_id: str = 'statements_too_open', display_name: str = 'Overly broad statements', category: str = 'cleanup')

Bases: object

Intelligence strategy: collect overly broad manage all-resources statements.

strategy_id: str = 'statements_too_open'

display_name: str = 'Overly broad statements'

category: str = 'cleanup'

run(repo: PolicyAnalysisRepository, overlay: dict, params: dict | None = None) → None

class oci_policy_analysis.application.core.engine.intelligence_strategies.cleanup_unused_dynamic_groups.UnusedDynamicGroupsCheck(strategy_id: str = 'unused_dynamic_groups', display_name: str = 'Unused dynamic groups', category: str = 'cleanup')

Bases: object

Intelligence strategy: collect unused dynamic groups for cleanup.

strategy_id: str = 'unused_dynamic_groups'

display_name: str = 'Unused dynamic groups'

category: str = 'cleanup'

run(repo: PolicyAnalysisRepository, overlay: dict, params: dict | None = None) → None

class oci_policy_analysis.application.core.engine.intelligence_strategies.cleanup_unused_groups.UnusedGroupsCheck(strategy_id: str = 'unused_groups', display_name: str = 'Unused groups', category: str = 'cleanup')

Bases: object

Intelligence strategy: collect groups with no users for cleanup.

strategy_id: str = 'unused_groups'

display_name: str = 'Unused groups'

category: str = 'cleanup'

run(repo: PolicyAnalysisRepository, overlay: dict, params: dict | None = None) → None

Consolidation

class oci_policy_analysis.application.core.engine.consolidation_engine.ConsolidationEngine(*, cache_mgr: CacheManager, reference_data_repo: ReferenceDataRepo, policy_repo: PolicyAnalysisRepository | None = None, strategies: list[Strategy] | None = None)

Bases: object

Middle-tier consolidation engine.

Instantiated early (UI startup) with cache and reference data. A PolicyAnalysisRepository is bound via bind_policy_repo() once the App creates it; the repository is populated later via tenancy load or cache load.

register_strategy(strategy: Strategy) → None

Register a single consolidation strategy (pluggable).

Parameters:

strategy – Strategy instance implementing the Strategy protocol.

register_strategies(strategies: list[Strategy]) → None

Register multiple consolidation strategies.

Parameters:

strategies – List of Strategy instances.

get_strategy_display_names() → list[str]

Return display names of all registered strategies (for UI dropdown, etc.).

Returns:

List of display_name strings in registration order (insertion order).

bind_policy_repo(repo: PolicyAnalysisRepository) → None

Attach the policy repository used for plan generation and rendering.

Parameters:

repo – The policy/compartment repository (e.g. policy_compartment_analysis).

detect_source_type() → Literal['live', 'cache', 'compliance', 'unknown']

Detect whether repo data is from live tenancy, cache, or compliance output.

Returns:

One of “live”, “cache”, “compliance”, or “unknown”.

tenancy_ocid() → str | None

Return tenancy OCID from the bound repo.

Returns:

Tenancy OCID string if repo is bound and has tenancy_ocid, else None.

dataset_version() → str | None

Return dataset version or data_as_of from the bound repo.

Returns:

Version string (e.g. timestamp) if repo has data_as_of, else None.

generate_plan(*, candidate_internal_ids: set[str], protected_internal_ids: set[str], strategy_display_name: str, params: dict[str, object] | None = None) → ConsolidationPlan

Build a consolidation plan for the given candidates and strategy.

Parameters:

• candidate_internal_ids – Set of statement internal_ids to consolidate.

• protected_internal_ids – Set of statement internal_ids to exclude from consolidation.

• strategy_display_name – Display name of the strategy (e.g. “Statement Density (Pack Policies)”).

• params – Optional strategy params (e.g. marker_tag_key).

Returns:

A ConsolidationPlan with plan_id, plan_steps, and plan_tags.

Raises:

• RuntimeError – If policy repo is not bound or tenancy_ocid is missing.

• ValueError – If strategy_display_name does not match any registered strategy.

render_plan_commands(plan: ConsolidationPlan) → str

Render CLI-oriented execution steps for the plan (no rollback hints).

Produces shell commands (oci iam policy update/delete) for administrators to run manually. Rollback hints for delete steps appear only in the rollback section (see render_plan_rollback_commands). The tool does not execute these commands.

Parameters:

plan – The consolidation plan to render.

Returns:

Multiline string of commented and executable CLI lines, or “(no repo bound)” if repo not set.

render_plan_ui_instructions(plan: ConsolidationPlan, section: Literal['all', 'execution', 'rollback'] = 'all') → str

Render step-by-step UI instructions for the OCI Console.

Produces human-readable steps for Identity & Security > Identity > Policies, including compartment navigation and location-change notes. Rollback block is included when section is “all” or “rollback”.

Parameters:

• plan – The consolidation plan to render.

• section – “execution” (steps only), “rollback” (rollback only), or “all” (both). Default “all”.

Returns:

Multiline string of instructions, or “(no repo bound)” if repo not set.

render_plan_rollback_commands(plan: ConsolidationPlan) → str

Render CLI rollback commands for the plan (reverse order; re-create for deletes).

Produces oci iam policy update/create commands to undo the plan. Shown in the Proposed Script area when the user selects “Rollback” or “Both”. Execution-only view does not include these.

Parameters:

plan – The consolidation plan to render rollback for.

Returns:

Multiline string of commented and executable CLI lines, or “(no repo bound)” if repo not set.

check_plan_progress(plan: ConsolidationPlan) → dict[str, dict[str, object]]

Inspect current repo state to estimate execution progress per step.

Uses policy presence (for delete steps) and marker tag presence (for modify steps) to determine which steps have been applied. Call after reloading policy data.

Parameters:

plan – The consolidation plan to check.

Returns:

Dict mapping step_id to a dict with keys such as executed (bool), notes (str), policy_ocid, action, checked_at.

Raises:

RuntimeError – If policy repository is not bound.

get_plan_tag_conflicts(plan: ConsolidationPlan, policies_by_ocid: dict[str, BasePolicy] | None = None) → list[dict[str, str]]

Check if any policy in this plan has an opa_consolidation (marker) tag from a different plan.

When multiple plans exist, a policy may already be tagged by another plan. If so, this plan is considered invalid/conflicted for execution until the situation is resolved. Works with any data source (live, cache, compliance); for static sources the result reflects the snapshot.

Parameters:

• plan – The consolidation plan to check.

• policies_by_ocid – Optional map policy_ocid -> policy; if None, built from bound repo.

Returns:

List of dicts with keys policy_ocid, current_tag_value, conflicting_plan_id. Empty if no conflicts.

oci_policy_analysis.application.core.common.consolidation_helpers.now_iso() → str

Return current UTC timestamp as ISO-8601 string.

oci_policy_analysis.application.core.common.consolidation_helpers.flatten_defined_tags(defined_tags: dict | None) → dict[str, str]

Flatten OCI defined tags into UI-friendly key space ‘namespace:key’. Not reversible; display only.

oci_policy_analysis.application.core.common.consolidation_helpers.policy_tag_maps(policy: BasePolicy) → tuple[dict[str, str], dict[str, dict[str, str]]]

Return (freeform_tags, defined_tags) as separate maps. Falls back to empty maps if absent.

oci_policy_analysis.application.core.common.consolidation_helpers.policy_statement_texts(repo: PolicyAnalysisRepository, policy_ocid: str) → list[str]

Return raw statement_text list for all regular statements in a policy.

oci_policy_analysis.application.core.common.consolidation_helpers.internal_id_to_statement(repo: PolicyAnalysisRepository) → dict[str, RegularPolicyStatement]

Build a map of statement internal_id -> statement dict for lookups.

oci_policy_analysis.application.core.common.consolidation_helpers.statement_scope_compartment_ocid(st: RegularPolicyStatement, tenancy_ocid: str) → str

OCI rule: statement cannot live in a policy below its scope. Returns compartment OCID for this statement’s scope.

oci_policy_analysis.application.core.common.consolidation_helpers.compartment_ancestors_including_self(compartment_ocid: str, compartments: list[dict]) → set[str]

Return set of compartment OCIDs: the given compartment and all ancestors up to root.

oci_policy_analysis.application.core.common.consolidation_helpers.lca_compartment_ocids(ocids: set[str], compartments: list[dict], tenancy_ocid: str) → str

Least common ancestor of the given compartment OCIDs. Policy containing these scopes must live here or an ancestor.

oci_policy_analysis.application.core.common.consolidation_helpers.normalize_compartment_path_segments(path: str) → list[str]

Split compartment path into segments (e.g. ROOT/A/B -> [‘ROOT’,’A’,’B’]). Preserves case.

oci_policy_analysis.application.core.common.consolidation_helpers.lca_path(paths: list[list[str]]) → list[str]

Find the longest common ancestor path among a list of path lists. For example, [[‘ROOT’,’A’,’B’], [‘ROOT’,’A’,’C’]] => [‘ROOT’,’A’] Returns the shared prefix as a list; returns empty list if none. Logs info about all paths and details of the comparison at each level.

oci_policy_analysis.application.core.common.consolidation_helpers.find_compartment_by_hierarchy_path(path: str, compartments: list[dict]) → dict | None

Looks up a compartment dict by case-insensitive match on hierarchy_path string. Example: path=’ROOT/department/foo’ Returns the compartment dict if found, else None.

oci_policy_analysis.application.core.common.consolidation_helpers.resolve_policy_compartment_path(policy: BasePolicy, compartments: list) → str

Return compartment hierarchy path for a policy: from policy.compartment_path or lookup by compartment_ocid.

oci_policy_analysis.application.core.common.consolidation_helpers.is_root_path(segments: list[str]) → bool

True if segments represent tenancy root (empty or single segment ROOT/root).

oci_policy_analysis.application.core.common.consolidation_helpers.effective_path_segments_for_rewrite(st: RegularPolicyStatement, old_location: str) → list[str]

Resolve effective path as segments from the statement. Prefer st.effective_path; else build from compartment_path + location.

oci_policy_analysis.application.core.common.consolidation_helpers.rewritten_location_for_target(effective_path: list[str], target_comp_path: list[str]) → str

Computes new location for a policy statement when moving to a new policy compartment.

Parameters:

• effective_path – List of strings, segments of the full effective path of the statement (e.g. [‘ROOT’,’LZ1-Top’,’app’]).

• target_comp_path – List of strings, segments of the target policy’s compartment path (e.g. [‘ROOT’,’LZ1-Top’])

Returns:

)

Return type:

A string representing the new location (the remainder after removing target_comp_path from effective_path, joined with

Notes: - No regex, no parsing of statement text is used. - If target_comp_path does not match the prefix of effective_path, returns original effective_path joined by ‘:’.

oci_policy_analysis.application.core.common.consolidation_helpers.rewrite_statement_location_clause(statement_text: str, new_location: str, *, target_policy_path: str) → tuple[str, str]

Rewrite in compartment <…> clause and return rewritten text + audit note.

If no location clause is found, the original text is returned unchanged while the note still documents the intended location rewrite.

oci_policy_analysis.application.core.common.consolidation_helpers.trace_and_rewrite_candidate_statement_location(*, strategy_id: str, internal_id: str, statement: RegularPolicyStatement, statement_text: str, target_policy_path: str) → tuple[str, str, list[str], list[str], str]

Trace location derivation for one candidate and return rewrite artifacts.

Returns:

(rewritten_text, note, effective_segments, target_segments, new_location)

Return type:

tuple

oci_policy_analysis.application.core.common.consolidation_helpers.required_policy_compartment_for_candidates(repo: PolicyAnalysisRepository, candidate_internal_ids: list[str], st_idx: dict[str, RegularPolicyStatement]) → str

Highest compartment in the tenancy where a policy can legally contain all candidate statements.