Module Development Guide

Module Development Guide

This guide reflects the current package layout and helper APIs.

Codebase Layout for Modules

gcpwn/modules/ is split into four top-level areas:

• gcpwn/modules/gcp/<service>/... — the GCP service modules (one dir per service: cloudscheduler, cloudworkflows, spanner, alloydb, orgpolicy, assetinventory, eventarc, workstations, billing, cloudshell, logging, memorystore, etc.).
• gcpwn/modules/workspace/... — Google Workspace modules (organized by API: cloud_identity/, directory/, with shared infra in workspace/common.py).
• gcpwn/modules/everything/... — cross-cutting modules that span services: enum_all, enum_gcp, enum_google_workspace, enum_gcp_policy_bindings, exploit_gcp_setiampolicy, process_gcp_iam_bindings, analyze_vulns.
• gcpwn/modules/opengraph/... — the BloodHound OpenGraph pipeline.

Within a service dir:

• Runtime modules: gcpwn/modules/gcp/<service>/<category>/...
• Service helpers: gcpwn/modules/gcp/<service>/utilities/helpers.py
• Static service data: gcpwn/modules/gcp/<service>/utilities/data/*
• Module index metadata: gcpwn/mappings/module_mappings.json

Namespace packages. Modules are now namespace packages — there is only a single __init__.py at gcpwn/__init__.py. Do not add empty __init__.py files under modules/. Refer to modules by their dotted path, e.g. gcpwn.modules.gcp.cloudscheduler.enumeration.enum_cloudscheduler.

The authoritative module list is gcpwn/mappings/module_mappings.json.

Minimum New Module Checklist

1. Add module file with run_module(user_args, session) entrypoint under the correct area (gcp/<service>/enumeration/... for a new service).
2. Use the declarative run_components pattern (below) for a standard list/get/testIamPermissions enumerator; fall back to in-module argparse (allow_abbrev=False) only for genuinely bespoke modules.
3. Use shared helpers from gcpwn/core/utils/service_runtime.py when possible.
4. Put the real API logic in utilities/helpers.py as a GcpListResource subclass; keep the module file a thin CLI wrapper.
5. Store discovered rows via the resource's save() (which calls session.insert_data(...)); record permissions/actions via the shared IAM/action helpers used by run_components.
6. Register the module in gcpwn/mappings/module_mappings.json and its table(s) in gcpwn/mappings/database_info.json.
7. If it should run under enum_all, wire it into gcpwn/modules/everything/enumeration/enum_all.py (see below).
8. Add tests (at least focused unit coverage where feasible).

The declarative enum pattern (preferred)

A standard service enumerator is now two small pieces:

1. A thin enum_<svc>.py that declares a list of Components and calls run_components.
2. A GcpListResource subclass in utilities/helpers.py that declares the table/columns/permissions and implements a couple of SDK hooks.

GcpListResource (gcpwn/core/resource.py) implements the shared list / get / test_iam_permissions / save bodies once. A subclass just declares config as class attributes and implements _build_client. Use gcp/cloudscheduler as the canonical template.

The helper (utilities/helpers.py)

from google.cloud import scheduler_v1

from gcpwn.core.resource import GcpListResource
from gcpwn.core.utils.module_helpers import (
    extract_path_segment,
    resolve_regions_from_module_data,
)


def resolve_locations(session, args) -> list[str]:
    return resolve_regions_from_module_data(
        session, args, service="cloudscheduler", discovery=("cloudscheduler", "v1")
    )


class CloudSchedulerJobsResource(GcpListResource):
    SERVICE_LABEL = "Cloud Scheduler"            # used in "API disabled" messages
    TABLE_NAME = "cloudscheduler_jobs"           # workspace-scoped DB table
    COLUMNS = ["location", "job_id", "name", "state", "schedule",
               "time_zone", "target_type", "target_uri", "target_sa_email"]
    ACTION_RESOURCE_TYPE = "jobs"
    LIST_PERMISSION = "cloudscheduler.jobs.list" # recorded as direct_api evidence on success
    GET_PERMISSION = "cloudscheduler.jobs.get"
    ID_FIELD = "job_id"                          # column for the short path tail

    def _build_client(self, session):
        return scheduler_v1.CloudSchedulerClient(credentials=session.credentials)

    def _list_items(self, parent, **_):
        return self.client.list_jobs(request=scheduler_v1.ListJobsRequest(parent=parent))

    def _get_item(self, resource_id, **_):
        return self.client.get_job(request=scheduler_v1.GetJobRequest(name=resource_id))

    def _extra_save_fields(self, raw):
        # derive extra columns from the raw resource (the offensively interesting bits)
        return {"job_id": extract_path_segment(str(raw.get("name", "") or ""), "jobs"), ...}

Key config attributes (full set documented in the GcpListResource docstring):

• SERVICE_LABEL, TABLE_NAME, COLUMNS, ACTION_RESOURCE_TYPE, ID_FIELD.
• LIST_PERMISSION / GET_PERMISSION — recorded as direct_api evidence on a successful list/get.
• TEST_IAM_API_NAME / TEST_IAM_PERMISSIONS — the testIamPermissions probe (recorded as test_iam_permissions evidence — distinct provenance).
• Parent-build mode: PARENT_FROM_PROJECT_LOCATION (default → projects/<p>/locations/<loc>), PARENT_FROM_PROJECT (→ projects/<p>, no location, e.g. pubsub/bigtable), or neither set (caller passes parent=).

Hooks you typically override:

• _build_client(session) — required; build the SDK client from session.credentials.
• _list_items(parent, **kwargs) / _get_item(resource_id, **kwargs) — override for clients that take request={...} / Request(...) objects or wrap responses (the default uses plain parent= / name= kwargs).
• _normalize_row(row) — post-process each resource_to_dict row.
• _extra_save_fields(raw) — derive extra columns to write alongside ID_FIELD (this is where you surface the offensively interesting fields, e.g. the SA a scheduler job impersonates).

DB writes are main-thread only. save() writes to the single-threaded DataController. Never call it from a parallel_map worker — run_components already collects on the main thread.

The module (enum_<svc>.py)

from gcpwn.core.utils.enum_framework import (
    REGION, Component, build_extra_args, component_args, run_components,
)
from gcpwn.core.utils.service_runtime import parse_component_args
from gcpwn.modules.gcp.cloudscheduler.utilities.helpers import (
    CloudSchedulerJobsResource, resolve_locations,
)

COMPONENTS = [
    Component("jobs", CloudSchedulerJobsResource, "Cloud Scheduler Jobs", "Jobs",
              help_text="Enumerate Cloud Scheduler jobs", scope=REGION,
              supports_iam=False,
              manual_id_arg="job_ids",
              manual_template=("projects", "{project_id}", "locations", 0, "jobs", 1),
              manual_help="Job IDs as LOCATION/JOB_ID or full projects/.../jobs/... names."),
]


def run_module(user_args, session):
    args = _parse_args(user_args)
    run_components(
        session, args, components=COMPONENTS,
        column_name="cloudscheduler_actions_allowed",
        region_resolver=resolve_locations, module_name="enum_cloudscheduler",
    )
    return 1

Each Component (gcpwn/core/utils/enum_framework.py) maps a CLI flag (--<key>) to a GcpListResource subclass plus knobs:

• scope: REGION (fan out list across regions), PROJECT (single global list), or NESTED (list under every parent component's rows — e.g. Spanner databases under instances; set parent_key).
• supports_iam=False — service has no per-resource testIamPermissions.
• persist=False — list/summarize only, never save() (resource has no table).
• summarize=False — module renders its own summary.
• manual_id_arg — enables --<arg> / --<arg>-file to target specific resources by name (hydrated via get(), so requires --get).

run_components handles the project/region fan-out, --get, --iam, per-row enrich_fn, saving, the summary, and flushing the three action accumulators once. Spanner (gcp/spanner) is a good --iam + NESTED (instances → databases) reference.

Module Shape (bespoke / non-declarative)

For modules that don't fit the list/get pattern, the contract is unchanged:

def run_module(user_args, session):
    parser = argparse.ArgumentParser(allow_abbrev=False, ...)
    # add flags
    args = parser.parse_args(user_args)
    # enumerate/list/get/testIamPermissions/download
    # write rows via session.insert_data(...)
    return 0   # truthy/0 == success; -1 == failure

Registering a new service

Three registries (plus optional enum_all wiring) must agree:

1. gcpwn/mappings/module_mappings.json — add the service block with its Enumeration category entry: module_name, info_blurb, and the dotted location (gcpwn.modules.gcp.<svc>.enumeration.enum_<svc>). Without this, modules list/info/run behavior is incomplete.
2. gcpwn/mappings/database_info.json — add a table_name entry under the database spec's tables array with columns (matching your resource's COLUMNS plus project_id, raw_json, and any derived fields) and primary_keys (typically ["project_id", "name"]). A workspace_id column + a workspaces(id) FK (ON DELETE CASCADE) and a (workspace_id, …) index are added automatically at table-build time — don't declare them yourself.
3. Tests — the contract tests (tests/module_contracts/) assert every mapped location exists and defines run_module.

Wiring a service into enum_all

gcpwn/modules/everything/enumeration/enum_all.py drives both a parallel orchestrator and a sequential path off one declarative _SERVICES ServiceSpec table. Adding a service touches four spots — all must get the new --flag:

1. _SERVICES — add a ServiceSpec row:

   ServiceSpec(("cloud_scheduler",),
               "gcpwn.modules.gcp.cloudscheduler.enumeration.enum_cloudscheduler",
               threads=True, regions=True),

   • gate_flags — the args attribute(s) that select this service (the --flag with dashes→underscores).
   • threads / regions / zones / iam / get — which standard args are forwarded. Use get=False for non-region custom modules that have no --get (e.g. cloud_billing, cloud_shell, cloud_logging).
   • opt_in=True — the service never runs under a bare enum_all; it runs only when its gate flag is passed. Cloud Asset Inventory (--asset-inventory) is the example: bare enum_all skips it, enum_all --asset-inventory runs everything plus it (additive).
2. _PARALLEL_SERVICES — add ("cloud_scheduler", "--cloud-scheduler") (the gate-key attr → its canonical flag name). This is now the single source of truth: the sequential parser's set_defaults, _ALL_GATE_KEYS, the derived every_flag_missing, the --modules token map, and the parallel orchestrator all read from it. You do not add a parser.add_argument or edit a hand-maintained every_flag_missing list anymore — a service added here runs under a bare enum_all and is selectable as --modules cloud-scheduler automatically. (Set opt_in=True on its ServiceSpec to keep it OUT of the bare-run sweep, like Cloud Asset Inventory.)
3. (Optional) _MODULE_TOKEN_ALIASES — add a friendly short token (e.g. "scheduler": "cloud_scheduler") if the full cloud-scheduler name is clunky.

Also keep the human-label map (e.g. "enum_cloudscheduler": "Cloud Scheduler") in sync so progress output and --list-modules are readable.

Conventions

• Prefer existing helper utilities over one-off implementations.
• Keep service API error handling consistent (403 denied, 404, API disabled, generic errors) via handle_service_error / handle_discovery_error — the GcpListResource bodies already route through these.
• Keep CLI output concise and table-aware; go through gcpwn.core.console.UtilityTools rather than hand-rolling ANSI.
• Respect workspace/project context and project selectors.
• Surface the offensively interesting field (the SA a job/trigger/workflow impersonates, a writer SA, an authorized SSH key) in _extra_save_fields.

Where command metadata comes from

Interactive module listing and info use:

• gcpwn/mappings/module_mappings.json

If you add a module but do not add this mapping, modules list/info/run behavior will be incomplete.

Testing

Run current suites:

python -m pytest -q -ra tests/unit tests/module_contracts

Relevant areas:

• tests/unit/core/
• tests/unit/opengraph/
• tests/unit/everything/
• tests/module_contracts/

OpenGraph-Specific Development

If adding new dangerous-path logic:

1. Update rule definitions (og_privilege_escalation_paths.json)
2. Update permission map coverage (og_permission_to_roles_map.json)
3. Add/adjust OpenGraph unit tests:
   • rule variants
   • combo path emission
   • coverage warnings and skip logic