##########################################################################
# Copyright (c) 2024, Oracle and/or its affiliates.
# Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl/
#
# DISCLAIMER This is not an official Oracle application, It does not supported by Oracle Support.
#
# caching.py
#
# @author: Andrew Gregory
#
# Supports Python 3.12 and above
#
# coding: utf-8
##########################################################################
import hashlib
import json
import os
import re
from collections.abc import Mapping, Sequence
from datetime import UTC, datetime
from pathlib import Path
from tempfile import NamedTemporaryFile
from typing import Any
from oci.identity.models import Domain
from oci_policy_analysis.application.core.repo import PolicyAnalysisRepository
from oci_policy_analysis.application.core.support.logger import get_logger
# Cache Directory and Date (for consistency across classes)
CACHE_DIR = Path.home() / '.oci-policy-analysis' / 'cache'
# AI result cache support has been removed (2025-11, per project guidance)
# Global logger for this module
logger = get_logger(component='caching')
CACHE_SCHEMA_VERSION = 3
[docs]
class CacheManager:
"""
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.
"""
@staticmethod
def _policy_statement_identity(item: Mapping[str, Any]) -> str:
"""Return identity key for policy statements, excluding mutable permission/action fields.
This is used so updates like verb/resource/action changes compare as
"Modified" rather than Added+Removed across snapshots.
"""
subject = item.get('subject')
if isinstance(subject, list):
subject_value = '|'.join(sorted(str(v) for v in subject))
else:
subject_value = str(subject or '')
return (
f'{item.get("policy_ocid") or item.get("policy_name") or ""}|'
f'{item.get("compartment_ocid") or item.get("compartment_path") or ""}|'
f'{item.get("subject_type") or ""}|{subject_value}|'
f'{item.get("location_type") or ""}|{item.get("location") or ""}|'
f'{item.get("conditions") or ""}|{item.get("comments") or ""}'
)
@staticmethod
def _stable_key_for_item(section: str, item: Mapping[str, Any]) -> str:
"""Return stable key used for historical diff normalization."""
if not isinstance(item, dict):
return ''
if section == 'policies':
return str(item.get('policy_ocid') or item.get('policy_name') or '')
if section == 'policy_statements':
identity = CacheManager._policy_statement_identity(item)
return hashlib.sha256(identity.encode('utf-8')).hexdigest()
if section == 'defined_aliases':
return (
f'{item.get("policy_ocid") or item.get("policy_name") or ""}|'
f'{item.get("defined_type") or ""}|{item.get("defined_name") or ""}|{item.get("ocid_alias") or ""}'
)
if section == 'cross_tenancy_statements':
return (
str(item.get('internal_id') or '')
or hashlib.sha256(
(
f'{item.get("policy_ocid") or item.get("policy_name") or ""}|{item.get("statement_text") or ""}'
).encode()
).hexdigest()
)
if section == 'identity_domains':
return str(item.get('id') or item.get('display_name') or item.get('name') or '')
if section == 'groups':
return str(item.get('group_ocid') or item.get('group_id') or '') or (
f'{item.get("domain_name") or ""}/{item.get("group_name") or ""}'
)
if section == 'dynamic_groups':
return str(item.get('dynamic_group_ocid') or item.get('dynamic_group_id') or '') or (
f'{item.get("domain_name") or ""}/{item.get("dynamic_group_name") or ""}'
)
if section == 'users':
return str(item.get('user_ocid') or item.get('user_id') or '') or (
f'{item.get("domain_name") or ""}/{item.get("user_name") or ""}'
)
return ''
def _annotate_stable_keys(self, section: str, rows: Sequence[Mapping[str, Any]]) -> list[dict[str, Any]]:
"""Annotate each row with stable_key (non-breaking additive field)."""
output: list[dict[str, Any]] = []
for row in rows or []:
if not isinstance(row, dict):
continue
row_copy = dict(row)
stable_key = self._stable_key_for_item(section, row_copy)
if stable_key:
row_copy['stable_key'] = stable_key
output.append(row_copy)
return output
@staticmethod
def _build_by_key(rows: Sequence[Mapping[str, Any]]) -> dict[str, dict[str, Any]]:
"""Build stable-key map representation for diff-friendly reads."""
by_key: dict[str, dict[str, Any]] = {}
for row in rows or []:
if not isinstance(row, dict):
continue
key = str(row.get('stable_key') or '')
if key:
by_key[key] = row
return by_key
# ----
# Canonical per-tenancy consolidation session file with protected_set & history
# ----
def _tenancy_state_path(self, category: str, tenancy_ocid: str) -> Path:
"""Use the same readable filename convention for all per-tenancy work."""
tenancy_key = str(tenancy_ocid or '')
if (
not tenancy_key
or tenancy_key.lower() in ('unknown', 'none')
or any(character in tenancy_key for character in ('/', '\\', '\0'))
or tenancy_key in ('.', '..')
):
raise ValueError(f'A valid tenancy OCID is required for {category} persistence.')
outdir = self.cache_dir / category
outdir.mkdir(parents=True, exist_ok=True)
return outdir / f'{category}_{tenancy_key}.json'
def _consolidation_state_path(self, tenancy_ocid):
"""Return path to per-tenancy consolidation state file."""
return self._tenancy_state_path('consolidation', tenancy_ocid)
def _cleanup_progress_path(self, tenancy_ocid: str) -> Path:
"""Keep mutable cleanup progress separate from immutable inventory snapshots."""
return self._tenancy_state_path('cleanup', tenancy_ocid)
def _legacy_cleanup_progress_path(self, tenancy_ocid: str) -> Path:
key = hashlib.sha256(str(tenancy_ocid).encode()).hexdigest()
return self.cache_dir / 'cleanup' / f'{key}.json'
[docs]
def load_cleanup_progress(self, tenancy_ocid: str) -> list[dict]:
path = self._cleanup_progress_path(tenancy_ocid)
if not path.exists():
path = self._legacy_cleanup_progress_path(tenancy_ocid)
if not path.exists():
return []
data = json.loads(path.read_text(encoding='utf-8'))
if data.get('tenancy_ocid') != tenancy_ocid or not isinstance(data.get('actions'), list):
raise ValueError('Cleanup progress does not match the requested tenancy.')
return [
dict(action)
for action in data['actions']
if isinstance(action, dict) and action.get('tenancy_ocid') == tenancy_ocid
]
[docs]
def save_cleanup_progress(self, tenancy_ocid: str, actions: list[dict]) -> None:
"""Atomically persist only this tenancy's tracked actions and verification history."""
import os
import tempfile
path = self._cleanup_progress_path(tenancy_ocid)
path.parent.mkdir(parents=True, exist_ok=True)
data = {
'version': 1,
'tenancy_ocid': tenancy_ocid,
'actions': [a for a in actions if a.get('tenancy_ocid') == tenancy_ocid],
}
fd, temporary = tempfile.mkstemp(dir=path.parent, suffix='.tmp')
try:
with os.fdopen(fd, 'w', encoding='utf-8') as handle:
json.dump(data, handle, indent=2)
os.replace(temporary, path)
# Only retire the legacy file once the canonical save succeeds.
self._legacy_cleanup_progress_path(tenancy_ocid).unlink(missing_ok=True)
finally:
if os.path.exists(temporary):
os.unlink(temporary)
[docs]
def read_consolidation_state(self, tenancy_ocid: str) -> dict:
"""Read saved consolidation state without creating or repairing a file.
Args:
tenancy_ocid: Saved tenancy identifier.
Returns:
dict: Existing state, with absent optional sections left absent.
Raises:
OSError: If the state file cannot be read.
ValueError: If JSON or section types are invalid.
"""
with self._consolidation_state_path(tenancy_ocid).open(encoding='utf-8') as source:
state = json.load(source)
if not isinstance(state, dict):
raise ValueError('Consolidation state must be a JSON object.')
if not isinstance(state.get('history') or [], list) or not isinstance(state.get('protected_set') or {}, dict):
raise ValueError('Consolidation history/protection has an invalid format.')
if not isinstance((state.get('protected_set') or {}).get('protected') or [], list):
raise ValueError('Protected statements must be a list.')
return state
[docs]
def get_or_create_consolidation_state(self, tenancy_ocid) -> dict:
"""
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.
"""
full_path = self._consolidation_state_path(tenancy_ocid)
if full_path.exists():
try:
with open(full_path, encoding='utf-8') as f:
data = json.load(f)
if 'protected_set' not in data:
data['protected_set'] = {}
if 'history' not in data:
data['history'] = []
return data
except Exception as e:
logger.error(f'Failed to load consolidation state for {tenancy_ocid}: {e}')
# Try to reinitialize state if corrupt
# New or failed to load: write default, return blank state
state = {'protected_set': {}, 'history': []}
with open(full_path, 'w', encoding='utf-8') as f:
json.dump(state, f, indent=2)
logger.info(f'Created new consolidation state file for tenancy_ocid={tenancy_ocid} at {full_path}')
return state
[docs]
def save_consolidation_state(self, tenancy_ocid, state: dict):
"""Overwrite the canonical consolidation state file for this tenancy_ocid."""
full_path = self._consolidation_state_path(tenancy_ocid)
with open(full_path, 'w', encoding='utf-8') as f:
json.dump(state, f, indent=2)
logger.info(f'Saved consolidation state for tenancy_ocid={tenancy_ocid} at {full_path}')
[docs]
def get_protected_set(self, tenancy_ocid):
"""Return the protected_set from consolidation state. Creates file if needed."""
state = self.get_or_create_consolidation_state(tenancy_ocid)
return state.get('protected_set', {})
[docs]
def set_protected_set(self, tenancy_ocid, protected_set: dict):
"""Update and persist the protected_set in the canonical consolidation state file."""
state = self.get_or_create_consolidation_state(tenancy_ocid)
state['protected_set'] = protected_set
self.save_consolidation_state(tenancy_ocid, state)
logger.info(f'Updated protected_set for tenancy_ocid={tenancy_ocid}')
[docs]
def get_history(self, tenancy_ocid):
"""
Return the list of run/plan history for this tenancy_ocid.
If no tenancy_ocid, returns empty list (safe for new sessions).
"""
if not tenancy_ocid or str(tenancy_ocid).lower() in ['unknown', '', 'none']:
logger.info('get_history: No tenancy_ocid provided, returning empty history list.')
return []
state = self.get_or_create_consolidation_state(tenancy_ocid)
return state.get('history', [])
[docs]
def add_run_record(self, tenancy_ocid, run_record: dict):
"""Append a new run/plan record to history and save the canonical file."""
state = self.get_or_create_consolidation_state(tenancy_ocid)
if 'history' not in state:
state['history'] = []
state['history'].append(run_record)
self.save_consolidation_state(tenancy_ocid, state)
logger.info(
f'Added run_record ({run_record.get("consolidation_effort_id", "unknown")}) for tenancy_ocid={tenancy_ocid}'
)
[docs]
def update_run_record(self, tenancy_ocid: str, consolidation_effort_id: str, updates: dict) -> bool:
"""
Update an existing run record in history by consolidation_effort_id.
Args:
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).
"""
if not tenancy_ocid or str(tenancy_ocid).lower() in ('unknown', '', 'none'):
return False
state = self.get_or_create_consolidation_state(tenancy_ocid)
history = state.get('history', [])
for i, run in enumerate(history):
if run.get('consolidation_effort_id') == consolidation_effort_id:
history[i] = {**run, **updates}
self.save_consolidation_state(tenancy_ocid, state)
logger.info(f'Updated run_record {consolidation_effort_id} for tenancy_ocid={tenancy_ocid}')
return True
return False
[docs]
def remove_run_record(self, tenancy_ocid: str, consolidation_effort_id: str) -> bool:
"""
Remove a run record from history by consolidation_effort_id.
Args:
tenancy_ocid: Tenancy OCID.
consolidation_effort_id: Run identifier (plan_id) to remove.
"""
if not tenancy_ocid or str(tenancy_ocid).lower() in ('unknown', '', 'none'):
return False
state = self.get_or_create_consolidation_state(tenancy_ocid)
history = state.get('history', [])
new_history = [r for r in history if r.get('consolidation_effort_id') != consolidation_effort_id]
if len(new_history) == len(history):
return False
state['history'] = new_history
self.save_consolidation_state(tenancy_ocid, state)
logger.info(f'Removed run_record {consolidation_effort_id} from tenancy_ocid={tenancy_ocid}')
return True
[docs]
def list_consolidation_tenancy_ocids(self) -> list[str]:
"""Return list of tenancy OCIDs that have consolidation state files."""
outdir = self.cache_dir / 'consolidation'
if not outdir.exists():
return []
ids = []
for f in outdir.glob('consolidation_*.json'):
name = f.stem
if name.startswith('consolidation_'):
ids.append(name[14:])
return ids
[docs]
def list_all_consolidation_plans(self) -> list[dict]:
"""
Return a flat list of all consolidation plans across tenancies.
Each item has tenancy_ocid, consolidation_effort_id, created_at, and run (full record).
"""
result = []
for tenancy_ocid in self.list_consolidation_tenancy_ocids():
history = self.get_history(tenancy_ocid)
for run in history:
result.append(
{
'tenancy_ocid': tenancy_ocid,
'consolidation_effort_id': run.get('consolidation_effort_id', ''),
'created_at': run.get('created_at', ''),
'run': run,
}
)
return result
def __init__(
self,
cache_dir: Path | None = None,
):
# logger = get_logger(component="caching")
self.cache_dir = Path(cache_dir).expanduser() if cache_dir else CACHE_DIR
self.cache_dir.mkdir(parents=True, exist_ok=True)
logger.info(f'Initialized Caching at {self.cache_dir}')
# ----
# Prospective statements (standalone per-tenancy storage)
# ----
def _prospects_path(self, tenancy_ocid: str) -> Path:
return self._tenancy_state_path('prospects', tenancy_ocid)
[docs]
def load_prospects(self, tenancy_ocid: str) -> list[dict[str, Any]]:
"""Load standalone prospective statements for a tenancy from cache/prospects."""
path = self._prospects_path(tenancy_ocid)
if not path.exists():
return []
try:
with open(path, encoding='utf-8') as f:
data = json.load(f)
if isinstance(data, list):
return [x for x in data if isinstance(x, dict)]
except Exception:
logger.warning('Failed loading prospects file %s', path, exc_info=True)
return []
[docs]
def save_prospects(self, tenancy_ocid: str, rows: list[dict[str, Any]]) -> None:
"""Persist standalone prospective statements for a tenancy to cache/prospects."""
path = self._prospects_path(tenancy_ocid)
with open(path, 'w', encoding='utf-8') as f:
json.dump(rows, f, ensure_ascii=False, indent=2, default=str)
logger.info('Saved %d prospective statements to %s', len(rows or []), path)
# ----
# Consolidation Session Save/Load
# ----
[docs]
def save_consolidation_session(
self, session_data: dict, plan_id: str | None = None, tenancy_ocid: str | None = None, preserved: bool = False
) -> str:
"""
Save a consolidation session (see models_consolidation.ConsolidationSession) to disk as JSON.
Returns file name/path.
Args:
session_data: The dict/TypedDict structure matching ConsolidationSession (pluggable, robust to version drift)
plan_id: Optional label/ID for file naming, else uses timestamp.
tenancy_ocid: Optional for nested cache structure.
preserved: Mark this session as not-to-be-deleted.
Returns:
The output file path string.
"""
# imports already present at the top
CACHE_DATE = datetime.now(UTC).strftime('%Y-%m-%d-%H-%M-%S-%Z')
tag = plan_id or f'session_{CACHE_DATE}'
tenancy_subdir = Path(str(tenancy_ocid)) if tenancy_ocid else Path()
outdir = self.cache_dir / 'consolidation' / tenancy_subdir
outdir.mkdir(parents=True, exist_ok=True)
file_path = outdir / f'{tag}.json'
def _serialize_for_json(obj):
if isinstance(obj, dict):
return {k: _serialize_for_json(v) for k, v in obj.items()}
elif isinstance(obj, list):
return [_serialize_for_json(x) for x in obj]
elif isinstance(obj, tuple):
return tuple(_serialize_for_json(x) for x in obj)
elif isinstance(obj, datetime):
return obj.isoformat()
else:
return obj
data = _serialize_for_json(session_data)
with open(file_path, 'w', encoding='utf-8') as f:
json.dump(data, f, ensure_ascii=False, indent=2)
logger.info(f'Saved consolidation session to {file_path}')
return str(file_path)
[docs]
def load_consolidation_session(self, plan_id: str, tenancy_ocid: str | None = None) -> dict | None:
"""
Load a consolidation session from cache.
Args:
plan_id: Label/filename (w/o json) for the session/plan.
tenancy_ocid: Optional subdir
Returns:
session_data (dict) or None
"""
# imports already present at the top
tenancy_subdir = Path(str(tenancy_ocid)) if tenancy_ocid else Path()
input_file = self.cache_dir / 'consolidation' / tenancy_subdir / f'{plan_id}.json'
try:
with open(input_file, encoding='utf-8') as f:
session_data = json.load(f)
logger.info(f'Loaded consolidation session from {input_file}')
return session_data
except Exception as e:
logger.error(f'Failed to load consolidation session {input_file}: {e}')
return None
# Utility functions for loading and saving cache, using combined caching strategy
[docs]
def save_combined_cache(
self, policy_analysis: PolicyAnalysisRepository, export_file=None, preserved: bool = False
) -> str:
"""
Save combined cache for policies and dynamic groups. Returns file name.
Args:
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"""
# Date of the cache
CACHE_DATE = datetime.now(UTC).strftime('%Y-%m-%d-%H-%M-%S-%Z')
# Supported snapshot structure: "policies" (BasePolicy objects), "policy_statements" (statements list)
policies = self._annotate_stable_keys('policies', policy_analysis.policies)
policy_statements = self._annotate_stable_keys('policy_statements', policy_analysis.regular_statements)
dynamic_groups = self._annotate_stable_keys('dynamic_groups', policy_analysis.dynamic_groups)
defined_aliases = self._annotate_stable_keys('defined_aliases', policy_analysis.defined_aliases)
cross_tenancy_statements = self._annotate_stable_keys(
'cross_tenancy_statements', policy_analysis.cross_tenancy_statements
)
identity_domains = self._annotate_stable_keys('identity_domains', policy_analysis._get_domains())
groups = self._annotate_stable_keys('groups', policy_analysis.groups)
users = self._annotate_stable_keys('users', policy_analysis.users)
combined_data = {
'version': 2,
'schema_version': CACHE_SCHEMA_VERSION,
'snapshot_id': hashlib.sha256(
f'{policy_analysis.tenancy_ocid}|{CACHE_DATE}|{policy_analysis.data_as_of}'.encode()
).hexdigest(),
'captured_at': datetime.now(UTC).isoformat(),
'generator_version': 'historical-schema-v3',
'tenancy_name': policy_analysis.tenancy_name,
'tenancy_ocid': policy_analysis.tenancy_ocid,
'policies': policies, # TODO(v4-removal): prefer policies_by_key consumers for diffing.
'policy_statements': policy_statements, # TODO(v4-removal): prefer policy_statements_by_key.
'defined_tag_namespace_keys': getattr(policy_analysis, 'defined_tag_namespace_keys', {}),
'dynamic_groups': dynamic_groups,
'defined_aliases': defined_aliases,
'cross_tenancy_statements': cross_tenancy_statements,
'compartments': policy_analysis.compartments,
'identity_domains': identity_domains,
'groups': groups,
'idp_group_mappings': getattr(policy_analysis, 'idp_group_mappings', []),
'users': users,
# New, additive map forms for stable keyed diffing.
'policies_by_key': self._build_by_key(policies),
'policy_statements_by_key': self._build_by_key(policy_statements),
'defined_aliases_by_key': self._build_by_key(defined_aliases),
'cross_tenancy_statements_by_key': self._build_by_key(cross_tenancy_statements),
'identity_domains_by_key': self._build_by_key(identity_domains),
'groups_by_key': self._build_by_key(groups),
'dynamic_groups_by_key': self._build_by_key(dynamic_groups),
'users_by_key': self._build_by_key(users),
'data_as_of': policy_analysis.data_as_of,
'load_all_users': getattr(policy_analysis, 'load_all_users', True),
'snapshot_kind': getattr(policy_analysis, 'snapshot_kind', None)
or ('policy_reload' if getattr(policy_analysis, 'policy_data_reloaded', None) else 'full'),
'recursive': getattr(policy_analysis, 'recursive', None),
'compartment_domain_search_depth': getattr(policy_analysis, 'compartment_domain_search_depth', None),
'compliance_capabilities': getattr(policy_analysis, 'compliance_capabilities', {}),
'inventory_complete': getattr(policy_analysis, 'inventory_complete', True)
and not bool(getattr(policy_analysis, '_cleanup_reload_api_errors', [])),
# Optional, tenancy/snapshot-specific metadata. Older cache files
# do not have this key and load normally.
'tenancy_policy_limits': getattr(policy_analysis, 'tenancy_policy_limits', {}),
}
logger.info(
'Saving combined cache: version=2, schema_version=3, "policies"=BasePolicy objects, "policy_statements"=statement list.'
)
def _serialize_for_json(obj):
"""Recursively convert datetime objects to ISO format (str)."""
if isinstance(obj, dict):
return {k: _serialize_for_json(v) for k, v in obj.items()}
elif isinstance(obj, list):
return [_serialize_for_json(x) for x in obj]
elif isinstance(obj, tuple):
return tuple(_serialize_for_json(x) for x in obj)
elif isinstance(obj, datetime):
return obj.isoformat()
else:
return obj
combined_data_serializable = _serialize_for_json(combined_data)
if export_file:
with open(export_file.name, 'w', newline='', encoding='utf-8') as filehandle:
json.dump(combined_data_serializable, filehandle, ensure_ascii=False, default=str)
logger.info(f'Exported combined cache to: {export_file.name}')
return str(export_file.name)
else:
combined_cache_file = self.cache_dir / f'combined_cache_{policy_analysis.tenancy_name}_{CACHE_DATE}.json'
with open(combined_cache_file, 'w', encoding='utf-8') as filehandle:
json.dump(combined_data_serializable, filehandle, ensure_ascii=False)
logger.info(f'Saved combined cache to: {combined_cache_file}')
# Update cache entries
entry = {
'tenancy_name': policy_analysis.tenancy_name,
'cache_date': CACHE_DATE,
'preserved': preserved,
}
entries_path = self.cache_dir / 'cache_entries.json'
with open(entries_path, 'a', encoding='utf-8') as date_file:
json.dump(entry, date_file, ensure_ascii=False)
date_file.write('\n') # Write a newline after each entry
logger.info(f'Updated cache entries with: {entry}')
# Cull old cache files and entries to keep only 10 most recent per tenancy
self._cull_old_caches(policy_analysis.tenancy_name)
# Return the name of the file
return str(combined_cache_file)
def _cull_old_caches(self, tenancy_name: str): # noqa: C901
"""Keep only the 10 most recent cache files/entries for this tenancy_name.
Preserved caches are never deleted."""
cache_files = list(self.cache_dir.glob(f'combined_cache_{tenancy_name}_*.json'))
# Gather preserved cache file names from entries
preserved_files = set()
entries_path = self.cache_dir / 'cache_entries.json'
if entries_path.exists():
with open(entries_path, encoding='utf-8') as f:
entry_lines = f.readlines()
for line in entry_lines:
try:
cache = json.loads(line)
if cache.get('tenancy_name') == tenancy_name and cache.get('preserved'):
preserved_files.add(f'combined_cache_{cache["tenancy_name"]}_{cache["cache_date"]}.json')
except Exception:
continue
# Only include files whose name matches date pattern
def parse_date_from_file(f):
# Example: combined_cache_andrew_2025-11-20-16-22-49-UTC.json
m = re.search(r'_(\d{4}-\d{2}-\d{2}-\d{2}-\d{2}-\d{2}-[A-Z]+)\.json$', f.name)
if m:
try:
return datetime.strptime(m.group(1), '%Y-%m-%d-%H-%M-%S-%Z')
except Exception:
return None
return None
dated_files = [(dt, f) for f in cache_files if (dt := parse_date_from_file(f)) is not None]
dated_files.sort(key=lambda x: x[0], reverse=True)
to_delete = [f for dt, f in dated_files[10:] if f.name not in preserved_files]
pruned = 0
# Do not delete preserved files or non-date-named
for old_file in to_delete:
try:
old_file.unlink()
logger.info(f'Pruned old cache file: {old_file}')
pruned += 1
except Exception as e:
logger.error(f'Could not remove old cache file {old_file}: {e}')
# Note: non-dated (renamed) caches are never automatically deleted here.
# Cull the cache_entries.json as well
if entries_path.exists():
with open(entries_path, encoding='utf-8') as f:
entry_lines = f.readlines()
remaining = []
count = 0
# Newest to oldest, keep up to 10 for tenancy_name, but never remove preserved
for line in reversed(entry_lines):
try:
cache = json.loads(line)
if cache.get('tenancy_name') == tenancy_name:
cache_path = (
self.cache_dir / f'combined_cache_{cache["tenancy_name"]}_{cache["cache_date"]}.json'
)
if cache.get('preserved', False) or (
cache_path.is_file() and parse_date_from_file(cache_path) is None
):
remaining.append(line)
elif count < 10:
remaining.append(line)
count += 1
# else skip (remove)
else:
remaining.append(line)
except Exception:
remaining.append(line) # keep malformed
# Write updated file (restore reversed order to maintain recency at top)
with open(entries_path, 'w', encoding='utf-8') as f:
for line in reversed(remaining):
f.write(line)
[docs]
def load_combined_cache(self, policy_analysis: PolicyAnalysisRepository, named_cache: str) -> str:
"""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.
Args:
named_cache: The tenancy_date string of the cache name to load
Returns:
A string indicating the name of the file used
"""
combined_cache_file = self.cache_dir / f'combined_cache_{named_cache}.json'
logger.info(f'Loading combined cache from: {combined_cache_file}')
if combined_cache_file.exists():
try:
with open(combined_cache_file, encoding='utf-8') as filehandle:
cache_data = json.load(filehandle)
# Grab all of the elements of the cache
# Required snapshot fields: "policies" and "policy_statements" must be present
if 'policies' not in cache_data or 'policy_statements' not in cache_data:
logger.error(
"Loaded cache file is missing required keys: 'policies' and/or 'policy_statements'."
)
raise RuntimeError(
'This cache file is not compatible with the current application version. '
'Please reload OCI data to generate a new cache file via the application UI.'
)
policy_analysis.policies = cache_data['policies']
policy_analysis.regular_statements = cache_data['policy_statements']
policy_analysis.defined_tag_namespace_keys = cache_data.get('defined_tag_namespace_keys', {})
dynamic_groups = cache_data.get('dynamic_groups', [])
cross_tenancy_data = cache_data.get('cross_tenancy_statements', [])
defined_aliases = cache_data.get('defined_aliases', [])
policy_analysis.tenancy_name = cache_data.get('tenancy_name', '')
policy_analysis.tenancy_ocid = cache_data.get('tenancy_ocid', '')
policy_analysis.compartments = cache_data.get('compartments', [])
policy_analysis.defined_aliases = defined_aliases
policy_analysis.cross_tenancy_statements = cross_tenancy_data
policy_analysis.dynamic_groups = dynamic_groups
policy_analysis.identity_domains = [
Domain(id=d['id'], display_name=d['display_name'], url=d['url'])
for d in cache_data.get('identity_domains', [])
]
policy_analysis.groups = cache_data.get('groups', [])
policy_analysis.idp_group_mappings = cache_data.get('idp_group_mappings', [])
policy_analysis.users = cache_data.get('users', [])
policy_analysis.version = cache_data.get('version', 1)
policy_analysis.load_all_users = cache_data.get('load_all_users', True)
for metadata_key in (
'snapshot_kind',
'recursive',
'compartment_domain_search_depth',
'compliance_capabilities',
'inventory_complete',
):
if metadata_key in cache_data:
setattr(policy_analysis, metadata_key, cache_data[metadata_key])
policy_analysis.tenancy_policy_limits = cache_data.get('tenancy_policy_limits', {}) or {}
policy_analysis.current_cache_name = named_cache
# Set the data as of time, always a str
policy_analysis.data_as_of = cache_data.get('data_as_of') or ''
# --- Load the last policy reload timestamp if present
policy_analysis.policy_data_reloaded = cache_data.get('policy_data_reloaded', '')
policy_analysis.enrich_display_structures()
logger.info(f'Loaded combined cache (strict mode) from: {combined_cache_file}')
logger.info(
f'Loaded {len(policy_analysis.policies)} BasePolicy objects, {len(dynamic_groups)} dynamic groups, '
f'{len(cross_tenancy_data)} cross-tenancy policies, '
f'{len(policy_analysis.identity_domains)} identity domains, '
f'{len(policy_analysis.groups)} groups, and {len(policy_analysis.users)} users from cache. '
f'Policy statements: {len(policy_analysis.regular_statements)}. '
f'Defined tag namespaces: {len(getattr(policy_analysis, "defined_tag_namespace_keys", {}) or {})}.'
)
except json.JSONDecodeError as e:
logger.error(f'Error decoding JSON from combined cache file: {e}')
return 'no cache'
except Exception as e:
logger.error(f'Error loading combined cache file: {e}')
return 'no cache'
# logger.warning(f'Unable to load data from cache: {combined_cache_file}')
else:
logger.warning(f'Unable to load data from cache: {combined_cache_file}')
raise ValueError('no cache')
return str(combined_cache_file)
[docs]
def load_cache_from_json(self, policy_analysis: PolicyAnalysisRepository, loaded_json: dict) -> bool:
"""
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.
Args:
loaded_json: The loaded JSON data as a dict
"""
try:
# Grab all of the elements of the cache
# Require both "policies" and "policy_statements" keys
if 'policies' not in loaded_json or 'policy_statements' not in loaded_json:
logger.error("Loaded cache (from JSON) missing required keys: 'policies' and/or 'policy_statements'.")
raise RuntimeError(
'This cache structure is incompatible with the current application version. '
'Please reload OCI data to create a new cache file.'
)
policy_analysis.policies = loaded_json['policies']
policy_analysis.regular_statements = loaded_json['policy_statements']
policy_analysis.defined_tag_namespace_keys = loaded_json.get('defined_tag_namespace_keys', {})
dynamic_groups = loaded_json.get('dynamic_groups', [])
cross_tenancy_data = loaded_json.get('cross_tenancy_statements', [])
defined_aliases = loaded_json.get('defined_aliases', [])
policy_analysis.tenancy_name = loaded_json.get('tenancy_name', '')
policy_analysis.tenancy_ocid = loaded_json.get('tenancy_ocid', '')
policy_analysis.compartments = loaded_json.get('compartments', [])
policy_analysis.defined_aliases = defined_aliases
policy_analysis.cross_tenancy_statements = cross_tenancy_data
policy_analysis.dynamic_groups = dynamic_groups
policy_analysis.identity_domains = [
Domain(id=d['id'], display_name=d['display_name'], url=d['url'])
for d in loaded_json.get('identity_domains', [])
]
policy_analysis.groups = loaded_json.get('groups', [])
policy_analysis.idp_group_mappings = loaded_json.get('idp_group_mappings', [])
policy_analysis.users = loaded_json.get('users', [])
policy_analysis.version = loaded_json.get('version', 1)
policy_analysis.load_all_users = loaded_json.get('load_all_users', True)
for metadata_key in (
'snapshot_kind',
'recursive',
'compartment_domain_search_depth',
'compliance_capabilities',
'inventory_complete',
):
if metadata_key in loaded_json:
setattr(policy_analysis, metadata_key, loaded_json[metadata_key])
policy_analysis.tenancy_policy_limits = loaded_json.get('tenancy_policy_limits', {}) or {}
policy_analysis.current_cache_name = ''
# Set the data as of time, always a str
policy_analysis.data_as_of = loaded_json.get('data_as_of') or ''
# --- Load last policy reload timestamp if present
policy_analysis.policy_data_reloaded = loaded_json.get('policy_data_reloaded', '')
policy_analysis.enrich_display_structures()
logger.info(
f'Loaded {len(policy_analysis.policies)} BasePolicy objects, {len(dynamic_groups)} dynamic groups, '
f'{len(cross_tenancy_data)} cross-tenancy policies, '
f'{len(policy_analysis.identity_domains)} identity domains, '
f'{len(policy_analysis.groups)} groups, and {len(policy_analysis.users)} users from cache (JSON input). '
f'Policy statements: {len(policy_analysis.regular_statements)}. '
f'Defined tag namespaces: {len(getattr(policy_analysis, "defined_tag_namespace_keys", {}) or {})}.'
)
return True
except json.JSONDecodeError as e:
logger.error(f'Error decoding JSON from combined cache file: {e}')
return False
except Exception as e:
logger.error(f'Error loading combined cache file: {e}')
return False
[docs]
def get_preserved_cache_set(self) -> set:
"""Get a set of cache names which are marked as preserved."""
preserved_files = set()
entries_path = self.cache_dir / 'cache_entries.json'
import json
if entries_path.exists():
with open(entries_path, encoding='utf-8') as f:
for line in f:
try:
entry = json.loads(line)
key = f'{entry["tenancy_name"]}_{entry["cache_date"]}'
if entry.get('preserved', False):
preserved_files.add(key)
except Exception:
continue
return preserved_files
[docs]
def get_available_cache(self, tenancy_name: str | None) -> list[str]:
"""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.
Args:
tenancy_name: The name of an OCI tenancy, which will filter the cache list down to only
caches for that tenancy.
Returns:
a list of the available named caches
"""
return_entries: list[str] = []
seen_entries: set[str] = set()
try:
with open(self.cache_dir / 'cache_entries.json', encoding='utf-8') as date_file:
entries = date_file.readlines()
logger.debug(f'Entries found in cache_entries.json: {entries}')
entries.reverse() # Show most recent first
logger.debug(f'Entries found in cache_entries.json (Reversed): {entries}')
for entry in entries:
cache = json.loads(entry)
if tenancy_name and cache['tenancy_name'] != tenancy_name:
continue
cache_name = cache['tenancy_name'] + '_' + cache['cache_date']
# Older desktop loads could append the same index entry twice
# within one second. Keep the cache usable but show it once.
if cache_name not in seen_entries:
return_entries.append(cache_name)
seen_entries.add(cache_name)
except json.JSONDecodeError:
logger.warning('No cache entries found or cache_entries.json is empty.')
except FileNotFoundError:
logger.warning('cache_entries.json file not found. No cache entries available.')
logger.info(f'Entries found in cache_entries.json: {len(return_entries)}')
return return_entries
[docs]
def load_cache_into_local_json(self, cached_tenancy: str) -> dict:
"""
Takes a named cache (tenancy_date) and returns the loaded JSON data as a dict.
Used for exporting or other purposes.
Args:
cached_tenancy: The tenancy_date string of the cache name to load
Returns:
The loaded cache data as a dict
"""
combined_cache_file = self.cache_dir / f'combined_cache_{cached_tenancy}.json'
if combined_cache_file.exists():
try:
with open(combined_cache_file, encoding='utf-8') as filehandle:
cache_data = json.load(filehandle)
logger.info(f'Loaded combined cache from: {combined_cache_file}')
# Show counts of each loaded element
# Return data as object
return cache_data
except json.JSONDecodeError as e:
logger.error(f'Error decoding JSON from combined cache file: {e}')
return {}
except Exception as e:
logger.error(f'Error loading combined cache file: {e}')
return {}
logger.warning(f'Unable to load data from cache: {combined_cache_file}')
return {}
[docs]
def update_tenancy_policy_limits(self, named_cache: str, limits: dict[str, Any]) -> bool:
"""Persist operator-supplied limits with one cache, never globally."""
if not named_cache:
return False
cache_file = self.cache_dir / f'combined_cache_{named_cache}.json'
try:
with open(cache_file, encoding='utf-8') as filehandle:
cache_data = json.load(filehandle)
cache_data['tenancy_policy_limits'] = limits
with open(cache_file, 'w', encoding='utf-8') as filehandle:
json.dump(cache_data, filehandle, ensure_ascii=False, indent=2, default=str)
return True
except (OSError, json.JSONDecodeError) as exc:
logger.warning('Could not save tenancy policy limits with cache %s: %s', named_cache, exc)
return False
@staticmethod
def _valid_cache_name(name: str) -> bool:
return (
isinstance(name, str)
and bool(name.strip())
and name == name.strip()
and '_' in name
and all(name.split('_', 1))
and not any(ord(char) < 32 or char in '<>:"/\\|?*' for char in name)
and not name.endswith(('.', ' '))
)
def _write_cache_index(self, lines: list[str]) -> None:
"""Atomically replace the local index, retaining malformed legacy lines."""
temporary = None
try:
with NamedTemporaryFile(
mode='w', encoding='utf-8', dir=self.cache_dir, prefix='.cache-index-', delete=False
) as stream:
temporary = Path(stream.name)
stream.writelines(lines)
temporary.replace(self.cache_dir / 'cache_entries.json')
finally:
if temporary is not None:
temporary.unlink(missing_ok=True)
@staticmethod
def _cache_index_name(line: str) -> str | None:
try:
entry = json.loads(line)
return f'{entry["tenancy_name"]}_{entry["cache_date"]}'
except (ValueError, KeyError, TypeError):
return None
[docs]
def remove_cache_entry(self, named_cache: str) -> bool:
"""Delete an unlocked cache and its index entry, preserving data on failure.
Args:
named_cache: Existing tenancy_label cache name; path components are rejected.
Returns:
bool: True after removal. Locked, missing or failed operations return False.
"""
if not self._valid_cache_name(named_cache):
return False
cache_file = self.cache_dir / f'combined_cache_{named_cache}.json'
entries = self.cache_dir / 'cache_entries.json'
original = None
changed = False
try:
original = entries.read_text(encoding='utf-8').splitlines(keepends=True)
matches = [line for line in original if self._cache_index_name(line) == named_cache]
if not cache_file.is_file() or not matches or any(json.loads(line).get('preserved') for line in matches):
return False
self._write_cache_index([line for line in original if self._cache_index_name(line) != named_cache])
changed = True
cache_file.unlink()
logger.info('Removed cache %s', named_cache)
return True
except OSError:
logger.exception('Cache deletion failed: %s', named_cache)
if changed and original is not None:
try:
self._write_cache_index(original)
except OSError:
logger.exception('Could not restore cache index; snapshot file is retained')
return False
[docs]
def rename_cache_entry(self, old_named_cache: str, new_named_cache: str) -> bool:
"""Rename an indexed cache without overwriting another snapshot.
Args:
old_named_cache: Existing tenancy_label cache name.
new_named_cache: New tenancy_label name with no path or Windows-invalid characters.
Returns:
bool: True after rename. Invalid names, collisions and failures return False.
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.
"""
if not all(self._valid_cache_name(name) for name in (old_named_cache, new_named_cache)):
return False
old_file = self.cache_dir / f'combined_cache_{old_named_cache}.json'
new_file = self.cache_dir / f'combined_cache_{new_named_cache}.json'
entries = self.cache_dir / 'cache_entries.json'
linked = changed = False
original = None
try:
original = entries.read_text(encoding='utf-8').splitlines(keepends=True)
names = [self._cache_index_name(line) for line in original]
if old_named_cache == new_named_cache:
return old_named_cache in names and old_file.is_file()
if old_named_cache not in names or new_named_cache in names or not old_file.is_file():
return False
updated = []
for line in original:
if self._cache_index_name(line) == old_named_cache:
entry = json.loads(line)
entry['tenancy_name'], entry['cache_date'] = new_named_cache.split('_', 1)
line = json.dumps(entry, ensure_ascii=False) + '\n'
updated.append(line)
os.link(old_file, new_file) # Exclusive creation; never overwrites a target.
linked = True
self._write_cache_index(updated)
changed = True
old_file.unlink()
logger.info('Renamed cache %s to %s', old_named_cache, new_named_cache)
return True
except OSError:
logger.exception('Cache rename failed: %s -> %s', old_named_cache, new_named_cache)
try:
if changed and original is not None:
self._write_cache_index(original)
if linked:
new_file.unlink(missing_ok=True)
except OSError:
logger.exception('Could not finish rename rollback; snapshot files are retained')
return False
[docs]
def preserve_cache_entry(self, named_cache: str, preserve: bool = True) -> bool:
"""Lock or unlock an existing cache against cleanup and manual deletion.
Args:
named_cache: Existing tenancy_label cache name.
preserve: True to lock, False to unlock.
Returns:
bool: True after the index is updated; False for missing or failed operations.
"""
if not self._valid_cache_name(named_cache):
return False
try:
entries = self.cache_dir / 'cache_entries.json'
original = entries.read_text(encoding='utf-8').splitlines(keepends=True)
updated, found = [], False
for line in original:
if self._cache_index_name(line) == named_cache:
entry = json.loads(line)
entry['preserved'] = preserve
line = json.dumps(entry, ensure_ascii=False) + '\n'
found = True
updated.append(line)
if not found or not (self.cache_dir / f'combined_cache_{named_cache}.json').is_file():
return False
self._write_cache_index(updated)
return True
except OSError:
logger.exception('Could not update cache lock: %s', named_cache)
return False
[docs]
def update_policy_section(
self,
policy_analysis: PolicyAnalysisRepository,
policy_data_reloaded: str | None,
*,
base_cache_name: str | None = None,
) -> str | None:
"""
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.
Args:
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.
"""
cache_files = list(self.cache_dir.glob(f'combined_cache_{policy_analysis.tenancy_name}_*.json'))
if not cache_files:
logger.warning(
f"[CacheManager.update_policy_section] No existing cache file for tenancy '{policy_analysis.tenancy_name}', skipping."
)
return None
def cache_name_for(path: Path) -> str:
return path.name.removeprefix('combined_cache_').removesuffix('.json')
available = {cache_name_for(path): path for path in cache_files}
source_name = str(base_cache_name or '')
source_file = available.get(source_name)
if source_file is None:
# A live tenancy load has no current_cache_name. Prefer the most
# recent full snapshot, so a reload never becomes the baseline by
# accident. Legacy caches without the marker are full snapshots.
full_caches = [path for path in cache_files if '_reload-' not in cache_name_for(path)]
source_file = max(full_caches or cache_files, key=lambda path: path.name)
source_name = cache_name_for(source_file)
root_name = re.sub(r'_reload-\d+$', '', source_name)
reload_number = 1
while f'{root_name}_reload-{reload_number:02d}' in available:
reload_number += 1
reload_name = f'{root_name}_reload-{reload_number:02d}'
cache_file = self.cache_dir / f'combined_cache_{reload_name}.json'
try:
with open(source_file, encoding='utf-8') as f:
cache_data = json.load(f)
except Exception as e:
logger.error(f"[CacheManager.update_policy_section] Failed to load cache file '{source_file}': {e}")
return None
# Update the compartment/policy section fields (other identity data are left untouched)
cache_data['policies'] = policy_analysis.policies
cache_data['policy_statements'] = policy_analysis.regular_statements
cache_data['defined_tag_namespace_keys'] = getattr(policy_analysis, 'defined_tag_namespace_keys', {})
cache_data['compartments'] = policy_analysis.compartments
cache_data['defined_aliases'] = policy_analysis.defined_aliases
cache_data['cross_tenancy_statements'] = policy_analysis.cross_tenancy_statements
cache_data['policies_by_key'] = self._build_by_key(
self._annotate_stable_keys('policies', policy_analysis.policies)
)
cache_data['policy_statements_by_key'] = self._build_by_key(
self._annotate_stable_keys('policy_statements', policy_analysis.regular_statements)
)
cache_data['defined_aliases_by_key'] = self._build_by_key(
self._annotate_stable_keys('defined_aliases', policy_analysis.defined_aliases)
)
cache_data['cross_tenancy_statements_by_key'] = self._build_by_key(
self._annotate_stable_keys('cross_tenancy_statements', policy_analysis.cross_tenancy_statements)
)
cache_data['policy_data_reloaded'] = policy_data_reloaded
cache_data['snapshot_kind'] = 'policy_reload'
cache_data['base_cache_name'] = root_name
cache_data['reload_source_cache_name'] = source_name
cache_data['identity_data_as_of'] = cache_data.get('data_as_of') or ''
cache_data['captured_at'] = datetime.now(UTC).isoformat()
cache_data['snapshot_id'] = hashlib.sha256(
f'{getattr(policy_analysis, "tenancy_ocid", "")}|{reload_name}|{policy_data_reloaded}'.encode()
).hexdigest()
try:
with open(cache_file, 'w', encoding='utf-8') as f:
json.dump(cache_data, f, ensure_ascii=False, indent=2, default=str)
entry = {
'tenancy_name': policy_analysis.tenancy_name,
'cache_date': reload_name.removeprefix(f'{policy_analysis.tenancy_name}_'),
'preserved': False,
'snapshot_kind': 'policy_reload',
'base_cache_name': root_name,
}
with open(self.cache_dir / 'cache_entries.json', 'a', encoding='utf-8') as entry_file:
json.dump(entry, entry_file, ensure_ascii=False)
entry_file.write('\n')
logger.info(
f'[CacheManager.update_policy_section] Created policy reload cache {cache_file} from {source_file}; policy_data_reloaded={policy_data_reloaded}'
)
return str(cache_file)
except Exception as e:
logger.error(f"[CacheManager.update_policy_section] Could not save cache file '{cache_file}': {e}")
return None