Module Development Guide

Module Development Guide

This guide reflects the current package layout and helper APIs. It captures the module contract and the shared runtime you should build on. For the canonical, in-repo version of these invariants see CLAUDE.md and CONTRIBUTING.md.

Codebase Layout for Modules

The installable package is nested at ocinferno/ocinferno/ (the repo root holds packaging/docs; the inner directory is the ocinferno package). Module code lives under ocinferno/modules/ in three areas:

• <service>/{enumeration,exploit,utilities}/ — the per-service modules (one directory per OCI service, e.g. dns/, objectstorage/, identityclient/). Categories are Enumeration (live reads), Process (operate on saved data), and Exploit (offensive write actions, currently under identityclient/exploit/). The enumeration file is a thin CLI wrapper; the real OCI API logic lives in utilities/helpers.py as Resource classes.
• everything/ — cross-cutting orchestrators: enum_all (cross-service enumeration) and process_config_check (the saved-data configuration audit engine, backed by utilities/config_audit.py).
• opengraph/ — the OpenGraph / BloodHound generator (process_oracle_cloud_hound_data). This is OCI-specific — do not force gcpwn parity here.

Modules are referred to by their dotted path following the folder layout, e.g. ocinferno.modules.dns.enumeration.enum_dns. The authoritative module list lives in ocinferno/mappings/module_mappings.json; the service-table schema lives in ocinferno/mappings/database_info.json.

The module contract

Every runnable module exposes a single entrypoint:

def run_module(user_args, session):
    ...
    return {"ok": True, "components": [ ... ]}

• user_args is a list[str] (argparse-style). Parse it with the helpers in core/utils/service_runtime.py (parse_wrapper_args, resolve_selected_components) rather than hand-rolling a new ArgumentParser.
• session is the SessionUtility. Its data API is:
  • session.save_resources(rows, table_name) — persist a list of row dicts. This is a single bulk executemany with one commit for the whole batch (unioning columns across rows so a field present on only some rows isn't dropped), so prefer passing the full list once over calling it row-by-row. Inside a DataController.transaction(...) it defers the commit to the enclosing transaction.
  • session.get_resource_fields(table, where_conditions=..., columns=...) — read saved rows back.
  • session.compartment_id — the compartment the current invocation is scoped to.
• The dispatcher wraps success/failure in an OperationResult, so the shape of the return value is not strictly enforced — but return the {"ok": ..., "components": [...]} dict for consistency.

Runner vs. module flags. The runner-level flags — --cids, --current-cid, --all-cids, --proxy, --get, --download, -v/--debug — are consumed in cli/module_actions.py and not forwarded verbatim; everything else on the line is module passthrough. --get/--download are re-surfaced to modules that opt into them. There is no --save/--no-save flag — enumeration always persists to the DB as it runs.

The declarative enum pattern (preferred)

A standard service enumerator is two small pieces:

1. A thin enum_<service>.py that declares its COMPONENTS/CACHE_TABLES and drives them through the shared runtime.
2. Resource classes in utilities/helpers.py that carry TABLE_NAME, COLUMNS, and list()/get()/save() (and optionally download()).

The helper (utilities/helpers.py)

Build the OCI client with the shared _init_client (it applies the workspace signer, proxy, and session behavior consistently), then implement list/get/save:

import oci
from ocinferno.core.utils.service_runtime import _init_client, ResourceBase


def build_dns_client(session, region=None):
    client = _init_client(oci.dns.DnsClient, session=session, service_name="DNS")
    target_region = region or getattr(session, "region", None)
    if target_region:
        client.base_client.set_region(target_region)
    return client


class DnsZonesResource(ResourceBase):
    TABLE_NAME = "dns_zones"                      # table name, registered in mappings/database_info.json
    COLUMNS = ["id", "name", "scope", "zone_type", "lifecycle_state", "time_created"]

    def __init__(self, session, region=None):
        self.session = session
        self.client = build_dns_client(session=session, region=region)

    def list(self, *, compartment_id):
        resp = oci.pagination.list_call_get_all_results(
            self.client.list_zones, compartment_id=compartment_id, scope="GLOBAL"
        )
        return oci.util.to_dict(resp.data) or []

    def get(self, *, resource_id):
        return oci.util.to_dict(self.client.get_zone(zone_name_or_id=resource_id).data) or {}

    # save() is inherited from ResourceBase (self.session.save_resources(rows, self.TABLE_NAME))
    # -- only override it if this resource needs custom save behavior.

Use oci.pagination.list_call_get_all_results for LIST calls and oci.util.to_dict(...) to normalize SDK models into plain dicts before saving. Extend ResourceBase (core/utils/service_runtime.py) rather than hand-rolling save() yourself -- it's the one-line save_resources(rows, TABLE_NAME) call every Resource class needs, plus a default no-op download() that services with binary artifacts override.

The module (enum_<service>.py)

Declare COMPONENTS as (key, method_suffix, help_text) tuples and a CACHE_TABLES map, parse with parse_wrapper_args, resolve which components were selected with resolve_selected_components, then run each selected component:

from ocinferno.core.console import UtilityTools
from ocinferno.core.utils.module_helpers import fill_missing_fields
from ocinferno.core.utils.service_runtime import (
    append_cached_component_counts, parse_wrapper_args, resolve_selected_components,
)
from ocinferno.modules.dns.utilities.helpers import DnsZonesResource

COMPONENTS = [
    ("zones", "zones", "Enumerate zones"),
]
CACHE_TABLES = {
    "zones": ("dns_zones", "compartment_id"),
}


def run_module(user_args, session):
    args, _ = parse_wrapper_args(
        user_args=user_args, description="Enumerate DNS resources", components=COMPONENTS,
    )
    component_order = [key for key, _s, _h in COMPONENTS]
    selected = resolve_selected_components(args, component_order)
    compartment_id = session.compartment_id

    results = []
    zones = DnsZonesResource(session=session)
    if selected.get("zones"):
        rows = [r for r in (zones.list(compartment_id=compartment_id) or []) if isinstance(r, dict)]
        for row in rows:
            row.setdefault("compartment_id", compartment_id)
        if args.get:
            for row in rows:
                fill_missing_fields(row, zones.get(resource_id=row["id"]) or {})
        if rows:
            UtilityTools.print_limited_table(rows, zones.COLUMNS)
        zones.save(rows)  # enumeration always persists -- there is no --save opt-in flag
        results.append({"ok": True, "zones": len(rows), "saved": True})

    append_cached_component_counts(
        results=results, session=session, selected=selected,
        component_order=component_order, cache_tables=CACHE_TABLES,
    )
    return {"ok": True, "components": results}

Notes on the shared runtime (core/utils/service_runtime.py):

• parse_wrapper_args(...) builds a --<component> flag per component plus the standard --get (and --download/--download-timeout when include_download=True), a --wide flag (print all columns instead of the concise default subset), and returns (args, remainder). There is no --save/--no-save flag: enumeration always persists to the DB as it runs (gcpwn parity, no opt-in step).
• resolve_selected_components(args, keys) returns {key: bool} — if the user named no component flags, all components run; otherwise only the named ones.
• append_cached_component_counts(...) reports counts for components the user did not select from their cached tables, and hints the exact modules run ... command to backfill missing prerequisite data.
• run_standard_enum_component(...) / ServiceEnumOpsBase package the list→get→print→save flow so a thin dispatcher can delegate instead of repeating the loop. ~18 modules now use it (grep ServiceEnumOpsBase under modules/ for current examples). The remaining hand-rolled modules mostly do so for a reason (content downloads, custom multi-scope listing, non-standard --get shape); migrating one is a deliberate, test-validated refactor, not a drop-in.
• A newer, more declarative alternative also exists: core/utils/enum_framework.py's Component/run_components/component_arg_specs (see modules/iot/enumeration/enum_iot.py or modules/lockbox/enumeration/enum_lockbox.py for real examples). It expresses each component as a Component(key, resource_cls, help_text=..., cache_table=...) entry and lets run_components(...) drive the whole list→get→print→save loop, including nested child-resource components (e.g. one resource fanned out per parent ID). Prefer this pattern for new modules with several related components; the hand-rolled COMPONENTS tuple style shown above is still fine for a single- or two-component module.

Registering a new service

Two registries must agree, plus module policy:

1. ocinferno/mappings/module_mappings.json — add a module entry:

   {
     "module_name": "enum_dns",
     "module_category": "Enumeration",
     "info_blurb": "Enumerate DNS resources",
     "author": "Your Name (@handle)",
     "location": "ocinferno.modules.dns.enumeration.enum_dns",
     "service": "DNS"
   }

   Without this, modules list/search/run behavior is incomplete.
2. ocinferno/mappings/database_info.json — add each new resource table under the top-level tables array with table_name, columns (matching your Resource's COLUMNS plus compartment_id, tag fields, and any derived columns), and primary_keys (typically ["id"]). A workspace_id column and the workspace-scoping index are added automatically by the schema builder — do not declare them yourself.
3. Module policy — behavior (run-once vs per-compartment, auth requirements) is decided by MODULE_POLICY_REGISTRY in cli/module_actions.py. The default (DEFAULT_MODULE_POLICY) is ExecMode.PER_TARGET (the module runs once per selected compartment), which is what a standard service enumerator wants. Only add an entry for a module that must be ExecMode.ONCE — e.g. enum_all and process_oracle_cloud_hound_data run a single time with a chosen context compartment rather than fanning out.

Critical invariants

1. One unified SQLite DB, workspace-scoped. All tables (workspaces / credentials / user-permissions plus every enumerated resource table, built from database_info.json) live in a single file, ocinferno/databases/ocinferno.db. metadata_db/service_db are just aliases to that one file, kept for API back-compat. Every service table carries a workspace_id FK (added automatically) — never query service tables without it. The session data API handles this for you.
2. Schema and registry are JSON. Tables come from database_info.json; modules from module_mappings.json. There is no YAML config — PyYAML is not a runtime dependency (the only YAML parsing is a lazy, optional fallback for OpenAPI specs in the API Gateway module).
3. Minimal required dependencies. Required runtime deps are oci, requests, prettytable, and oci-lexer-parser. The Excel-export stack (pandas, xlsxwriter) is the optional [excel] extra and is lazily imported — keep it that way so a base install stays lean.
4. Errors go through the result contract. Use OperationResult.from_exception / UtilityTools.dlog (see core/contracts.py and core/console.py) for normalized error handling rather than re-implementing per-module status/code extraction.
5. OpenGraph is OCI-specific. The modules/opengraph/ logic (policy lexing via oci-lexer-parser, dynamic-group matching, identity-domain graph building) is intentionally different from GCPwn's OpenGraph. Leave the graph-building logic alone unless explicitly asked.

Conventions

• Output goes through core.console.UtilityTools (colors, print_limited_table, dlog, OCID condensing). Don't hand-roll ANSI codes.
• Normalize SDK models to dicts with oci.util.to_dict(...) before saving.
• Surface the offensively interesting field (a policy statement, an attached dynamic group, a writable bucket, an authorized key) in the saved columns.
• Keep non-OCI runtime dependencies minimal (eases enterprise install approval).

Testing

python -m pytest -q -ra tests/unit          # primary suite (run in CI)
ruff check .                                 # lint (BLOCKING in CI)
mypy                                         # type-check core (informational)

Install dev tooling with pip install .[dev]. Excel-related tests skip unless the optional extra is installed (pip install .[excel]). Additional suites live under tests/integration/ and tests/enum_modules/.