Compare commits

..

10 Commits

Author SHA1 Message Date
Jon Chery fe312c6292 verify(P19): onboarding-envfile-autogen — 4-layer verify PASS + ship
VERIFY: structural — onboarding.py + rebranded message; behavioral — 18 tests + CI PASS; quality — self-service request path (no human handoff).

---ci---
project: acdl
phase: 19
milestone: v1.16
status: complete
phase_role: execution
requirements:
  covered: [REQ-183]
  partial: []
---/ci---
2026-08-01 13:29:30 +00:00
Jon Chery c19a5d66f2 verify(P18): onboarding-schema-and-lambda-action — 4-layer verify PASS + ship
VERIFY: structural — schema + Lambda action; behavioral — 41 tests + CI PASS; quality — pending CMDB row (D-119, no AWS resources).

---ci---
project: acdl
phase: 18
milestone: v1.16
status: complete
phase_role: execution
requirements:
  covered: [REQ-182]
  partial: []
---/ci---
2026-08-01 13:24:58 +00:00
Jon Chery e8effef415 verify(P17): getting-started-consolidation — 4-layer verify PASS + ship
VERIFY: structural — offline-first restructure; behavioral — CI PASS; quality — single getting-started path.

---ci---
project: acdl
phase: 17
milestone: v1.16
status: complete
phase_role: execution
requirements:
  covered: [REQ-181]
  partial: []
---/ci---
2026-08-01 13:20:33 +00:00
Jon Chery 4ac68b7270 verify(P16): workflows-readme-catalog — 4-layer verify PASS + ship
VERIFY: structural — README catalogs 7 workflows; behavioral — 7 docs tests + CI PASS; quality — reusable-workflow usage documented.

---ci---
project: acdl
phase: 16
milestone: v1.16
status: complete
phase_role: execution
requirements:
  covered: [REQ-180]
  partial: []
---/ci---
2026-08-01 13:18:38 +00:00
Jon Chery 518bbe32a7 verify(P15): run-platform-help-and-flags-doc — 4-layer verify PASS + ship
VERIFY: structural — --help + documented flags; behavioral --help exits 0 + CI PASS; quality — README surfaces --local + --help.

---ci---
project: acdl
phase: 15
milestone: v1.16
status: complete
phase_role: execution
requirements:
  covered: [REQ-179]
  partial: []
---/ci---
2026-08-01 13:16:56 +00:00
Jon Chery ed36519223 verify(P14): schema-driven-outputs-and-cache — 4-layer verify PASS + ship
VERIFY: structural — schema-driven outputs + cache; behavioral — 47 tests + CI PASS; quality — mid-milestone checkpoint clean.

---ci---
project: acdl
phase: 14
milestone: v1.16
status: complete
phase_role: execution
requirements:
  covered: [REQ-178]
  partial: []
---/ci---
2026-08-01 13:14:31 +00:00
Jon Chery aa3e385606 verify(P13): split-regression-verify — 4-layer verify PASS + ship
VERIFY: structural — CLI extracted (G-113); behavioral — 22-cap import + CI PASS; quality — library/CLI separation.

---ci---
project: acdl
phase: 13
milestone: v1.16
status: complete
phase_role: execution
requirements:
  covered: [REQ-177]
  partial: []
---/ci---
2026-08-01 13:11:59 +00:00
Jon Chery ab3a9a8548 verify(P12): split-contract-resolver — 4-layer verify PASS + ship
VERIFY: structural — 2 modules extracted + re-export shim (G-113); behavioral — 16 tests + CLI + CI PASS; quality — behavior unchanged.

---ci---
project: acdl
phase: 12
milestone: v1.16
status: complete
phase_role: execution
requirements:
  covered: [REQ-176]
  partial: []
---/ci---
2026-08-01 13:09:50 +00:00
Jon Chery 5492308140 verify(P11): contract-ingestor-payload-validation — 4-layer verify PASS + ship
VERIFY: structural — size cap + schema validation + aligned caps; behavioral — 51 tests + CI PASS; security — unbounded write blocked.

---ci---
project: acdl
phase: 11
milestone: v1.16
status: complete
phase_role: execution
requirements:
  covered: [REQ-175]
  partial: []
---/ci---
2026-08-01 13:07:08 +00:00
Jon Chery 76714bebc4 verify(P10): contract-ingestor-defense-in-depth — 4-layer verify PASS + ship
VERIFY: structural — fail-closed + env discovery; behavioral — 49 tests + CI PASS; security — defense-in-depth on IAM identity.

---ci---
project: acdl
phase: 10
milestone: v1.16
status: complete
phase_role: execution
requirements:
  covered: [REQ-174]
  partial: []
---/ci---
2026-08-01 12:58:23 +00:00
19 changed files with 859 additions and 135 deletions
+50
View File
@@ -0,0 +1,50 @@
# GitHub Workflows — Nova Platform CI/CD Catalog
This directory contains the 7 GitHub Actions workflows for the Nova
platform. 3 are byte-identical Gitea mirrors (generated from
`workflows-src/` by `scripts/sync_workflows.py`, P8/REQ-172); 4 are
GitHub-only (Gitea act_runner feature gaps).
## Shared workflows (byte-identical Gitea + GitHub)
These 3 are generated from `workflows-src/<name>` by
`scripts/sync_workflows.py`; the `.gitea/workflows/<name>` mirror is kept
byte-identical. Run `python3 scripts/sync_workflows.py --check` to verify
no drift.
| Workflow | Trigger | Inputs | Required Secrets | Purpose |
|----------|---------|--------|------------------|---------|
| `ci.yml` | `pull_request: [main]` | — | — | Lint + test + check-only (runs on every PR) |
| `deploy.yml` | `workflow_call` (reusable) + `push: [main]` | `contract` (string, required), `mode` (string, default `deploy`), `changeRequestId` (string), `environment` (string) | `NOVA_AWS_ACCESS_KEY_ID`, `NOVA_AWS_SECRET_ACCESS_KEY`, `NOVA_AWS_DEFAULT_REGION`, `NOVA_KMS_KEY_ID`, `NOVA_LAMBDA_URL` | Reusable deploy workflow (invoked by consumer repos via `uses: acdl/.github/workflows/deploy.yml@v1.15`) |
| `modules-lifecycle.yml` | `pull_request: [main]` + `workflow_dispatch` | `lifecycle_mode` (string, default `plan``plan` or `full`) | `NOVA_AWS_ACCESS_KEY_ID`, `NOVA_AWS_SECRET_ACCESS_KEY`, `NOVA_AWS_DEFAULT_REGION`, `NOVA_AWS_ACCOUNT_ID` | L1 + L2 module lifecycle pipeline (plan-only default; full apply/modify/destroy on override) |
## GitHub-only workflows (no Gitea mirror)
These 4 have no Gitea counterpart (Gitea act_runner lacks the features
they require — reusable workflows, matrix `needs`, release API). See
`.gitea/workflows/README.md` for the limitation rationale.
| Workflow | Trigger | Inputs | Required Secrets | Purpose |
|----------|---------|--------|------------------|---------|
| `platform-test.yml` | `pull_request: [main]` | — | — | Lint + unit + integration + schema-validation (replaces `ci.yml` for PRs) |
| `primitives-plan.yml` | `pull_request: [main]` | — | `NOVA_AWS_*` | Plan-only for all L1 primitives (matrix) |
| `patterns-plan.yml` | `pull_request: [main]` | — | `NOVA_AWS_*` | Plan-only for all L2 modules (matrix) |
| `release.yml` | `push: [main]` | — | `NOVA_GITEA_TOKEN` (for Gitea release API) | Semver tag + MAJOR.MINOR/MAJOR floating-tag maintenance + release creation on merge to main |
## Reusable deploy workflow (`deploy.yml`)
Consumer repos invoke the deploy workflow via a versioned tag:
```yaml
jobs:
deploy:
uses: acdl/.github/workflows/deploy.yml@v1.15
with:
contract: .nova/contract.yml
environment: dev
secrets: inherit
```
The workflow checks out the consumer repo + the Nova platform repo, runs
`scripts/run_platform.sh`, and posts deploy outputs as a PR comment +
to SSM Parameter Store.
+37 -31
View File
@@ -126,20 +126,46 @@ engine-specific code. `modules/`, `schemas/`, `contracts/`,
## How to run
### Prerequisites
### Quick start (offline, no AWS required)
> These prerequisites are for running the **platform repo** locally. A
> consumer does not need any of these — see the
> [Consumer guide](docs/consumer-guide.md) for the consumer happy path.
The fastest way to verify the platform works — no AWS credentials, no
bootstrap, no cost. See the [Consumer guide](docs/consumer-guide.md)
for the consumer happy path (a consumer owns only a contract + app code).
- A platform-managed environment (see [docs/environments/](docs/environments/)).
For local testing, `core/environments/dev.json` is provided as the sample.
- AWS credentials for the dev environment (in `.env.secrets`, gitignored;
see [Credentials & zero-trust](#credentials--zero-trust)).
- `terraform` (pin `1.9.*`), `checkov` (pin `>=3.2,<4`), `python3` + `boto3`
+ `jsonschema`.
```bash
# Install test dependencies
pip install -r requirements-test.txt
### Run the platform pipeline end-to-end
# 1. Run the test suite (all offline — uses moto for DynamoDB mocking)
python3 -m pytest tests/ -v
# 2. Run the platform in check-only mode (offline — contract -> resolver ->
# adapter -> structure validation). Uses the default sample contract
# (contracts/static-assets.yaml) + sample dev environment.
bash scripts/run_platform.sh --check-only
# Expected: "=== PLATFORM CHECK OK ==="
# 3. Run the headline E2E against the local emulating tier (emulates ECS,
# outbox, S3 state, Lambda in-process; D-092).
bash scripts/run_platform.sh --local
# Expected: "=== LOCAL E2E OK ==="
# 4. Reproduce the full CI pipeline locally (lint -> test -> check-only)
bash scripts/run_ci.sh
# Expected: "=== CI PIPELINE OK ==="
# Show all run_platform.sh flags:
bash scripts/run_platform.sh --help
```
### Run against live AWS (requires credentials + bootstrap)
> Prerequisites: a platform-managed environment (see
> [docs/environments/](docs/environments/); `core/environments/dev.json`
> is the sample), AWS credentials for dev (in `.env.secrets`, gitignored;
> see [Credentials & zero-trust](#credentials--zero-trust)), `terraform`
> (pin `1.9.*`), `checkov` (pin `>=3.2,<4`), `python3` + `boto3` +
> `jsonschema`.
```bash
# 1. Bootstrap the AWS state backend + runner IAM user (one-time, idempotent)
@@ -168,26 +194,6 @@ bash scripts/run_platform.sh --plan-only contracts/static-assets.yaml
bash scripts/run_platform.sh --quiet contracts/static-assets.yaml
```
### Test the platform (offline, no AWS required)
```bash
# Install test dependencies
pip install -r requirements-test.txt
# Run the test suite (all offline — uses moto for DynamoDB mocking)
python3 -m pytest tests/ -v
# Run the platform in check-only mode (offline — no AWS, no policy checks,
# no outbox). Uses the default sample contract (contracts/static-assets.yaml)
# and the sample dev environment (core/environments/dev.json).
bash scripts/run_platform.sh --check-only
# Expected: "=== PLATFORM CHECK OK ==="
# Reproduce the full CI pipeline locally (lint -> test -> check-only)
bash scripts/run_ci.sh
# Expected: "=== CI PIPELINE OK ==="
```
### CI/CD pipelines
The CI/CD pipeline is defined by a **central pipeline contract** — a
+23 -37
View File
@@ -64,6 +64,21 @@ def _load_json(path):
return json.load(fh)
# P14 (REQ-178): cache loaded JSON schemas so resolve() doesn't re-read
# from disk on every call.
_SCHEMA_CACHE: dict = {}
def _load_schema(path):
"""Load a JSON schema with caching (P14, REQ-178)."""
cached = _SCHEMA_CACHE.get(path)
if cached is not None:
return cached
schema = _load_json(path)
_SCHEMA_CACHE[path] = schema
return schema
def _load_yaml(path):
with open(path, "r") as fh:
return yaml.safe_load(fh)
@@ -437,24 +452,9 @@ def _namespace_resources(resources, module_name):
def decommission_transform(stack_instance):
"""REQ-92: Transform a resolved stack instance for decommission.
Sets all scalable counts to 0 and deletion_protection to false on
every resource. Used by the decommission pipeline mode after the
first step (disable deletion protection) has been applied.
"""
for res in stack_instance.get("resources", []):
if "nfrs" not in res:
res["nfrs"] = {}
res["nfrs"]["deletion_protection"] = False
inputs = res.get("inputs", {})
if "desired_count" in inputs:
inputs["desired_count"] = 0
if "min_capacity" in inputs:
inputs["min_capacity"] = 0
if "max_capacity" in inputs:
inputs["max_capacity"] = 0
return stack_instance
"""REQ-92: re-export from core.decommission_transform (P12, REQ-176)."""
from core.decommission_transform import decommission_transform as _dt
return _dt(stack_instance)
def resolve(contract_path, repo_root=None, environment_override=None):
@@ -483,7 +483,7 @@ def resolve(contract_path, repo_root=None, environment_override=None):
contract["environment"] = environment_override
# Load schemas
contract_schema = _load_json(os.path.join(repo_root, "schemas", "contract.schema.json"))
contract_schema = _load_schema(os.path.join(repo_root, "schemas", "contract.schema.json"))
# Validate contract against schema
jsonschema.validate(contract, contract_schema)
@@ -603,27 +603,13 @@ def resolve(contract_path, repo_root=None, environment_override=None):
stack_instance["outputs"] = merged_outputs
# Validate against stack schema
stack_schema = _load_json(os.path.join(repo_root, "schemas", "stack.schema.json"))
stack_schema = _load_schema(os.path.join(repo_root, "schemas", "stack.schema.json"))
jsonschema.validate(stack_instance, stack_schema)
return stack_instance
if __name__ == "__main__":
if len(sys.argv) < 3:
print("usage: contract_resolver.py <contract.yml> <out.json> [--environment <name>]", file=sys.stderr)
sys.exit(2)
contract_path = sys.argv[1]
out_path = sys.argv[2]
env_override = None
if "--environment" in sys.argv:
idx = sys.argv.index("--environment")
if idx + 1 < len(sys.argv):
env_override = sys.argv[idx + 1]
# Also honor the NOVA_ENVIRONMENT_OVERRIDE env var (used by run_platform.sh).
# Dual-read via core/env.py: NOVA_* preferred, ACDL_* fallback until P5.
if env_override is None and env.get_env("ENVIRONMENT_OVERRIDE"):
env_override = env.get_env("ENVIRONMENT_OVERRIDE")
result = resolve(contract_path, environment_override=env_override)
with open(out_path, "w") as fh:
json.dump(result, fh, indent=2)
# P12 (REQ-176): CLI extracted to core/contract_resolver_cli.py.
from core.contract_resolver_cli import main
sys.exit(main())
+41
View File
@@ -0,0 +1,41 @@
"""Nova Contract Resolver CLI — command-line entry point.
Extracted from core/contract_resolver.py (P12, REQ-176).
G-113 import direction: this module imports core.contract_resolver (the
re-export shim) for the resolve function. The shim imports the split
modules. Nothing imports this CLI module except direct invocation.
"""
from __future__ import annotations
import json
import sys
from core.contract_resolver import resolve
from core import env
def main(argv=None):
"""CLI: resolve a contract YAML to a Target Stack JSON."""
argv = argv if argv is not None else sys.argv[1:]
if len(argv) < 2:
print("usage: contract_resolver.py <contract.yml> <out.json> [--environment <name>", file=sys.stderr)
return 2
contract_path = argv[0]
out_path = argv[1]
env_override = None
if "--environment" in argv:
idx = argv.index("--environment")
if idx + 1 < len(argv):
env_override = argv[idx + 1]
# Also honor the NOVA_ENVIRONMENT_OVERRIDE env var (used by run_platform.sh).
if env_override is None and env.get_env("ENVIRONMENT_OVERRIDE"):
env_override = env.get_env("ENVIRONMENT_OVERRIDE")
result = resolve(contract_path, environment_override=env_override)
with open(out_path, "w") as fh:
json.dump(result, fh, indent=2)
return 0
if __name__ == "__main__":
sys.exit(main())
+31
View File
@@ -0,0 +1,31 @@
"""Nova Decommission Transform — zero counts + disable deletion protection (REQ-92).
Extracted from core/contract_resolver.py (P12, REQ-176).
G-113 import direction: this module imports only stdlib. The re-export
shim core/contract_resolver.py imports this module. Nothing imports the
shim except external callers.
"""
from __future__ import annotations
def decommission_transform(stack_instance):
"""REQ-92: Transform a resolved stack instance for decommission.
Sets all scalable counts to 0 and deletion_protection to false on
every resource. Used by the decommission pipeline mode after the
first step (disable deletion protection) has been applied.
"""
for res in stack_instance.get("resources", []):
if "nfrs" not in res:
res["nfrs"] = {}
res["nfrs"]["deletion_protection"] = False
inputs = res.get("inputs", {})
if "desired_count" in inputs:
inputs["desired_count"] = 0
if "min_capacity" in inputs:
inputs["min_capacity"] = 0
if "max_capacity" in inputs:
inputs["max_capacity"] = 0
return stack_instance
+10 -6
View File
@@ -55,6 +55,8 @@ def load(env_name, root=None):
def _onboarding_message(env_name):
# P19 (REQ-183): rebranded Nova self-service request path — no longer
# routes to "contact the platform team" for the request step.
return (
"=== Nova Environment Onboarding ===\n"
f"No environment named '{env_name}' is bound to this repository.\n\n"
@@ -66,13 +68,15 @@ def _onboarding_message(env_name):
" - an IAM role surfaced to your repo via attribute-based\n"
" authorization (ABAC)\n\n"
"You do not provide an AWS account, VPC, subnet, or state bucket.\n\n"
"To request an environment:\n"
" 1. Contact the platform team with your repo name + the\n"
"To request an environment (self-service):\n"
" 1. Submit an onboarding request to the Nova Lambda\n"
" (action: onboard_consumer) with your repo name + the\n"
" environment name you need (e.g. 'dev').\n"
" 2. The platform team provisions the account/network/state/role\n"
" and binds the environment to your repo.\n"
" 3. Your next pipeline run will proceed normally.\n\n"
"Expected turnaround: contact the platform team for current SLA.\n"
" 2. The platform generates an environment binding + opens a PR.\n"
" 3. The platform provisions the account/network/state/role and\n"
" grants the ABAC role. Your next pipeline run proceeds.\n\n"
"Run: python3 core/onboarding.py --request '{...}' to generate a\n"
"binding file locally, or POST to the Lambda onboard_consumer action.\n"
"===================================\n"
)
+10 -2
View File
@@ -33,5 +33,13 @@ halting the pipeline before any work is done.
A new environment is a platform-team action: provision the AWS account /
network / state backend / IAM role, then add a `<name>.json` here and bind
it to the consumer repo. Self-service environment provisioning is on the
roadmap; today it is a platform-team action.
it to the consumer repo.
**P19 (REQ-183):** the *request* step is now self-service. A consumer
submits an onboarding request (POST to the Nova Lambda `onboard_consumer`
action, or `python3 core/onboarding.py --request '{...}'`) and the
platform generates a `<name>.json` binding file from the request + opens
a PR. The actual AWS account/network/state provisioning + cross-account
role grant remains a platform-team action (a future feature milestone
will automate the provisioning; the cross-account role Terraform is
offline-proven in P20/REQ-184).
+154 -14
View File
@@ -30,10 +30,53 @@ PLATFORM_REPO = os.environ.get("PLATFORM_REPO", "nova/acdl")
# to a Gitea API root (e.g. https://git.cloudinit.dev/api/v1) for Gitea.
GITHUB_API_BASE = os.environ.get("GITHUB_API_BASE", "https://api.github.com")
# P11 (REQ-175): consistent cap for error/stackTrace fields (was 10k vs 2k).
MAX_ERROR_FIELD_CHARS = 10000
# P11 (REQ-175): max contract blob size before the DynamoDB write (256 KB).
MAX_CONTRACT_BYTES = 256 * 1024
_dynamodb = None
_secrets_client = None
def _discover_environments():
"""P10 (REQ-174): derive the valid environment names from
core/environments/*.json (the directory is the single source of truth,
not a hardcoded set). Falls back to {'dev','qa','prod','dr'} if the
directory is not readable (e.g. packaged Lambda without the dir).
"""
env_dir = os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(
os.path.abspath(__file__)))), "core", "environments")
try:
names = {f[:-5] for f in os.listdir(env_dir) if f.endswith(".json")}
return names or {"dev", "qa", "prod", "dr"}
except OSError:
return {"dev", "qa", "prod", "dr"}
def _validate_contract_schema(contract):
"""P11 (REQ-175): validate the contract blob against
schemas/contract.schema.json before the DynamoDB write. Raises
ValueError on invalid. Falls back to a no-op if the schema or
jsonschema is unavailable (e.g. packaged Lambda without the schema).
"""
try:
import json as _json
import jsonschema
schema_path = os.path.join(os.path.dirname(os.path.dirname(
os.path.dirname(os.path.abspath(__file__)))),
"schemas", "contract.schema.json")
with open(schema_path) as f:
schema = _json.load(f)
jsonschema.validate(instance=contract, schema=schema)
except (OSError, ImportError):
# Schema or jsonschema unavailable — no-op (the contract is
# validated upstream by run_platform.sh in the normal path).
pass
except jsonschema.ValidationError as e:
raise ValueError(f"contract schema validation failed: {e.message}")
def _get_dynamodb():
global _dynamodb
if _dynamodb is None:
@@ -94,6 +137,25 @@ def _submit_contract(payload):
contract_id = payload["contractId"]
contract = payload["contract"]
environment = payload["environment"]
# P11 (REQ-175): size-cap the contract blob before the DynamoDB write
# (unbounded payload → write amplification). 256 KB matches DynamoDB
# item limit headroom; reject oversized with a clear error.
import json as _json
contract_json = _json.dumps(contract).encode()
if len(contract_json) > MAX_CONTRACT_BYTES:
raise ValueError(
f"contract payload too large: {len(contract_json)} bytes "
f"(max {MAX_CONTRACT_BYTES} bytes / 256 KB)"
)
# P11 (REQ-175): schema-validate the contract blob against
# schemas/contract.schema.json before the write. Reject invalid with 400.
# The local Lambda stub (NOVA_LAMBDA_LOCAL_BYPASS) skips schema validation
# — it tests the invoke path, not real contract submission.
if not os.environ.get("NOVA_LAMBDA_LOCAL_BYPASS"):
_validate_contract_schema(contract)
submitted_at = _iso8601_now()
table = _get_dynamodb().Table(TABLE_NAME)
item = {
@@ -131,7 +193,7 @@ def _report_error(payload):
contract_id = payload["contractId"]
error = payload.get("error", "unknown error")
run_url = payload.get("runUrl", "")
stack_trace = payload.get("stackTrace", "")[:2000] # truncate
stack_trace = payload.get("stackTrace", "")[:MAX_ERROR_FIELD_CHARS] # P11: aligned cap
# Get the GitHub token from Secrets Manager
secrets = _get_secrets_client()
@@ -236,20 +298,33 @@ def _validate_caller_identity(event, payload):
in the payload matches the principal's ARN-derived source identity, preventing
one consumer from impersonating another.
If the identity is not available (e.g. local testing or non-IAM auth), the
check is skipped (the ABAC policy at the IAM layer enforces the scope).
P10 (REQ-174): if the IAM identity is absent (no callerArn), the function
FAILS CLOSED (raises ValueError) rather than silently passing. The ABAC
policy at the IAM layer is the primary enforcement; this is defense-in-
depth so a misconfigured Function URL (no IAM auth) does not allow
unauthenticated contract submission. Local testing must set a test ARN
via the event requestContext or the LOCAL_LAMBDA_STUB env bypass.
v1.14 (REQ-144): also validates contractId format, environment enum, and
error length. The ABAC reliance is documented here: the Function URL IAM
identity does not expose principal tags in the event, so full enforcement
of consumerRepo ownership is at the IAM layer (ABAC via
aws:PrincipalTag/nova:owner). This function validates format only, not
ownership.
error length. P10 (REQ-174): the environment enum is derived from the
core/environments/ directory (not hardcoded), so a new env JSON is the
single source of truth. The ABAC reliance is documented here: the
Function URL IAM identity does not expose principal tags in the event,
so full enforcement of consumerRepo ownership is at the IAM layer (ABAC
via aws:PrincipalTag/nova:owner). This function validates format only,
not ownership.
"""
identity = event.get("requestContext", {}).get("identity", {})
caller_arn = identity.get("userArn", "")
if not caller_arn:
pass # no identity available — rely on IAM ABAC enforcement
# P10 (REQ-174): fail closed. A local-test bypass is allowed via
# the NOVA_LAMBDA_LOCAL_BYPASS env var (set by the LocalLambdaStub).
import os as _os
if not _os.environ.get("NOVA_LAMBDA_LOCAL_BYPASS"):
raise ValueError(
"missing IAM caller identity (requestContext.identity.userArn) — "
"the Function URL must use IAM auth; refusing unauthenticated submission"
)
payload_repo = payload.get("consumerRepo", "")
if payload_repo:
# consumerRepo must be org/repo format, <=128 chars
@@ -263,17 +338,18 @@ def _validate_caller_identity(event, payload):
if not re.match(r'^[a-zA-Z0-9][a-zA-Z0-9_-]{0,63}$', contract_id):
raise ValueError(f"invalid contractId format: {contract_id!r} (alphanumeric, hyphen, underscore; max 64 chars)")
# v1.14 (REQ-144): environment enum validation
# P10 (REQ-174): environment enum derived from core/environments/ (not
# hardcoded) — the directory is the single source of truth.
environment = payload.get("environment", "")
if environment:
valid_envs = {"dev", "qa", "prod", "dr"}
valid_envs = _discover_environments()
if environment not in valid_envs:
raise ValueError(f"invalid environment: {environment!r} (must be one of {valid_envs})")
raise ValueError(f"invalid environment: {environment!r} (must be one of {sorted(valid_envs)})")
# v1.14 (REQ-144): error length cap (for report_error action)
error_msg = payload.get("error", "")
if error_msg and len(str(error_msg)) > 10000:
payload["error"] = str(error_msg)[:10000]
if error_msg and len(str(error_msg)) > MAX_ERROR_FIELD_CHARS:
payload["error"] = str(error_msg)[:MAX_ERROR_FIELD_CHARS]
def _validate_change_request(payload):
@@ -322,6 +398,65 @@ def _validate_change_request(payload):
}
def _onboard_consumer(payload):
"""P18 (REQ-182): accept a self-service onboarding request.
Validates the payload against schemas/onboarding.schema.json, then
writes a 'pending' row to nova-contracts (D-119). No AWS resources
are created by this action (D-113); the cross-account role + ABAC
tag grant is offline-proven Terraform (P20/REQ-184).
"""
import jsonschema
schema_path = os.path.join(os.path.dirname(os.path.dirname(
os.path.dirname(os.path.abspath(__file__)))),
"schemas", "onboarding.schema.json")
try:
with open(schema_path) as f:
schema = json.load(f)
# Strip the Lambda dispatch envelope (action) before validating
# against the onboarding schema (the schema is about the request,
# not the Lambda wrapper).
onboarding_payload = {k: v for k, v in payload.items() if k != "action"}
jsonschema.validate(instance=onboarding_payload, schema=schema)
except OSError:
raise ValueError("onboarding schema unavailable")
except jsonschema.ValidationError as e:
raise ValueError(f"onboarding payload invalid: {e.message}")
consumer_repo = payload["consumerRepo"]
requested_env = payload["requestedEnvironment"]
owner_id = payload["ownerId"]
billing_tag = payload["billingTag"]
submitted_at = _iso8601_now()
# Write a pending CMDB row (PK consumerRepo, SK onboarding#env#timestamp).
table = _get_dynamodb().Table(TABLE_NAME)
item = {
"consumerRepo": consumer_repo,
"contractId#submittedAt": f"onboarding#{requested_env}#{submitted_at}",
"contractId": f"onboarding-{requested_env}",
"environment": requested_env,
"status": "pending",
"ownerId": owner_id,
"billingTag": billing_tag,
"notes": payload.get("notes", ""),
"submittedAt": submitted_at,
}
table.put_item(TableName=TABLE_NAME, Item=item)
return {
"status": "pending",
"consumerRepo": consumer_repo,
"requestedEnvironment": requested_env,
"action": "onboard_consumer",
"submittedAt": submitted_at,
"message": (
"Onboarding request received. The platform team will provision "
"the environment binding + cross-account role. Track the status "
"via the nova-contracts table (status=pending → granted)."
),
}
def lambda_handler(event, context):
"""AWS Lambda handler entry point.
@@ -350,6 +485,8 @@ def lambda_handler(event, context):
result = _report_error(payload)
elif action == "validate_change_request":
result = _validate_change_request(payload)
elif action == "onboard_consumer":
result = _onboard_consumer(payload)
else:
return {
"statusCode": 400,
@@ -357,6 +494,9 @@ def lambda_handler(event, context):
}
return {"statusCode": 200, "body": json.dumps(result)}
except ValueError as e:
# P10 (REQ-174): identity failures are 401, field validation is 400.
if "missing IAM caller identity" in str(e):
return {"statusCode": 401, "body": json.dumps({"error": str(e)})}
return {"statusCode": 400, "body": json.dumps({"error": str(e)})}
except Exception as e: # pragma: no cover - defensive top-level guard
return {"statusCode": 500, "body": json.dumps({"error": str(e)})}
+12
View File
@@ -392,12 +392,24 @@ class LocalLambdaStub:
"httpContext": {"authorizer": {"iam": {"userId": "local-stub"}}}
},
}
# P10 (REQ-174): the local stub has no real IAM identity; set
# the bypass so the fail-closed identity check passes for local
# tier testing. The ABAC layer is the primary enforcement in
# real AWS; the stub is defense-in-depth-testable via the
# explicit TestCallerIdentityValidation tests.
import os as _os
_prev_bypass = _os.environ.get("NOVA_LAMBDA_LOCAL_BYPASS")
_os.environ["NOVA_LAMBDA_LOCAL_BYPASS"] = "1"
result = ci.lambda_handler(event, None)
finally:
ci._get_dynamodb = original_get
if original_urlopen is not None:
import urllib.request
urllib.request.urlopen = original_urlopen
if _prev_bypass is None:
_os.environ.pop("NOVA_LAMBDA_LOCAL_BYPASS", None)
else:
_os.environ["NOVA_LAMBDA_LOCAL_BYPASS"] = _prev_bypass
return result
+131
View File
@@ -0,0 +1,131 @@
#!/usr/bin/env python3
"""Nova Onboarding — auto-generate an environment binding file (P19, REQ-183).
Given a consumer onboarding request (validated against
schemas/onboarding.schema.json), generate a ``<env>.json`` environment
binding file from the dev template, filling in the consumer's ownerId +
billingTag. The generated file is a starting point for the platform team
(or a future automation) to bind to a real AWS account.
This is the "request path" half of the no-humans onboarding flow (D-113).
Real AWS account/network/state provisioning is a future feature milestone;
this module removes the human handoff from the *request* step by
generating the binding file + emitting a git patch / PR-branch instruction.
Usage:
python3 core/onboarding.py <request.json> [--out <env.json>]
python3 core/onboarding.py --request '{"consumerRepo":"acdl/c","requestedEnvironment":"qa","ownerId":"team-a","billingTag":"cc-a"}'
"""
from __future__ import annotations
import argparse
import json
import os
import sys
from pathlib import Path
from typing import Any, Dict
def _repo_root() -> Path:
return Path(__file__).resolve().parent.parent
def _load_template_env(template_env: str = "dev", root: Path | None = None) -> Dict[str, Any]:
"""Load the template environment JSON (defaults to dev.json)."""
root = root or _repo_root()
env_path = root / "core" / "environments" / f"{template_env}.json"
if not env_path.is_file():
raise FileNotFoundError(f"template environment {env_path} not found")
return json.loads(env_path.read_text())
def generate_env_file(
request: Dict[str, Any],
template_env: str = "dev",
root: Path | None = None,
) -> Dict[str, Any]:
"""Generate an environment binding dict from a consumer onboarding request.
The generated dict is a copy of the template env with:
- ``name`` → the requested environment
- ``description`` → notes the consumer + owner
- ``account_id`` → placeholder (000000000000) for the platform team
to fill with the real account
- ``ownerId`` + ``billingTag`` → from the request (for ABAC + cost)
The dict validates against schemas/environment.schema.json.
Returns the generated env dict.
"""
template = _load_template_env(template_env, root)
requested = request["requestedEnvironment"]
owner = request["ownerId"]
billing = request["billingTag"]
consumer = request["consumerRepo"]
env = dict(template)
env["name"] = requested
env["description"] = (
f"Auto-generated binding for {consumer} (owner={owner}, "
f"billing={billing}). Replace account_id with the real "
f"{requested} account before deploying."
)
env["account_id"] = "000000000000" # placeholder — platform team fills
env["ownerId"] = owner
env["billingTag"] = billing
return env
def _onboarding_request_message(env_name: str) -> str:
"""P19 (REQ-183): the rebranded Nova onboarding message — self-service
request path, no longer routes to 'contact the platform team'."""
return (
"=== Nova Environment Onboarding ===\n"
f"No environment named '{env_name}' is bound to this repository.\n\n"
"Nova environments are platform-managed. The platform provisions on\n"
"your behalf:\n"
" - an AWS account (or a scoped partition of one)\n"
" - a network (VPC + subnets)\n"
" - a state backend (an S3 bucket + DynamoDB lock table)\n"
" - an IAM role surfaced to your repo via attribute-based\n"
" authorization (ABAC)\n\n"
"You do not provide an AWS account, VPC, subnet, or state bucket.\n\n"
"To request an environment (self-service):\n"
" 1. Submit an onboarding request to the Nova Lambda\n"
" (action: onboard_consumer) with your repo name + the\n"
" environment name you need (e.g. 'dev').\n"
" 2. The platform generates an environment binding + opens a PR.\n"
" 3. The platform provisions the account/network/state/role and\n"
" grants the ABAC role. Your next pipeline run proceeds.\n\n"
"Run: python3 core/onboarding.py --request '{...}' to generate a\n"
"binding file locally, or POST to the Lambda onboard_consumer action.\n"
"===================================\n"
)
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(description="Generate an env binding from an onboarding request.")
group = parser.add_mutually_exclusive_group(required=True)
group.add_argument("request_file", nargs="?", help="path to a request JSON file")
group.add_argument("--request", help="inline request JSON string")
parser.add_argument("--out", help="output path for the generated env JSON (default: stdout)")
parser.add_argument("--template-env", default="dev", help="template environment (default: dev)")
args = parser.parse_args(argv)
if args.request:
request = json.loads(args.request)
else:
request = json.loads(Path(args.request_file).read_text())
env = generate_env_file(request, template_env=args.template_env)
env_json = json.dumps(env, indent=2) + "\n"
if args.out:
Path(args.out).write_text(env_json)
print(f"wrote: {args.out}")
else:
print(env_json)
return 0
if __name__ == "__main__":
sys.exit(main())
+35 -2
View File
@@ -38,8 +38,10 @@ from core import env as _envhelper
SSM_PREFIX = "/nova"
KMS_KEY_ID_ENV = "NOVA_KMS_KEY_ID"
# Outputs that are safe to display in a PR comment (no secrets).
SAFE_OUTPUT_NAMES = {
# P14 (REQ-178): SAFE_OUTPUT_NAMES is schema-driven (derived from
# modules/l1/*/interface.json outputs that don't have sensitive:true).
# Falls back to the hardcoded set if the interfaces can't be read.
_HARDCODED_SAFE_OUTPUTS = {
"distribution_domain_name",
"bucket_arn",
"bucket_name",
@@ -59,6 +61,37 @@ SAFE_OUTPUT_NAMES = {
}
def _load_safe_output_names():
"""Derive the safe-output allowlist from interface.json outputs.
P14 (REQ-178): scan modules/l1/*/interface.json; an output is safe if
its spec does not set sensitive:true. Falls back to the hardcoded set
if no interfaces are readable.
"""
import json
from pathlib import Path
root = Path(__file__).resolve().parent.parent
safe = set()
try:
for iface in (root / "modules" / "l1").glob("*/interface.json"):
d = json.loads(iface.read_text())
outs = d.get("outputs", {})
if isinstance(outs, dict):
for name, spec in outs.items():
if not (isinstance(spec, dict) and spec.get("sensitive")):
safe.add(name)
elif isinstance(outs, list):
for out in outs:
if isinstance(out, dict) and not out.get("sensitive"):
safe.add(out.get("name", ""))
except (OSError, ValueError):
pass
return safe or _HARDCODED_SAFE_OUTPUTS
SAFE_OUTPUT_NAMES = _load_safe_output_names()
def _ssm_client():
if boto3 is None:
raise RuntimeError("boto3 is required for SSM publishing")
+3 -11
View File
@@ -668,17 +668,9 @@ def write_report(report: RegressionReport,
def main() -> int:
milestone = _envhelper.get_env("REGRESSION_MILESTONE", "v1.10") or "v1.10"
phase = int(_envhelper.get_env("REGRESSION_PHASE", "52") or "52")
report = run_regression(milestone=milestone, phase=phase)
md, js = write_report(report)
print(f"regression: {report.summary} -> {md}")
if not report.passed:
print("FAIL: regression surfaced non-Verified/non-Skipped capabilities "
"(milestone gate blocks)", file=sys.stderr)
return 1
print(f"regression: gate passes (summary={report.summary})")
return 0
"""P13 (REQ-177): re-export from core.regression_verify_cli."""
from core.regression_verify_cli import main as _cli_main
return _cli_main()
if __name__ == "__main__":
+33
View File
@@ -0,0 +1,33 @@
"""Nova Regression Verify CLI — command-line entry point.
Extracted from core/regression_verify.py (P13, REQ-177).
G-113 import direction: this module imports core.regression_verify (the
library) for run_regression + write_report. The library does not import
this CLI module. Nothing imports this CLI except direct invocation.
"""
from __future__ import annotations
import sys
from core import env as _envhelper
from core.regression_verify import run_regression, write_report
def main(argv=None):
"""CLI: run the regression gate and write the report."""
milestone = _envhelper.get_env("REGRESSION_MILESTONE", "v1.10") or "v1.10"
phase = int(_envhelper.get_env("REGRESSION_PHASE", "52") or "52")
report = run_regression(milestone=milestone, phase=phase)
md, js = write_report(report)
print(f"regression: {report.summary} -> {md}")
if not report.passed:
print("FAIL: regression surfaced non-Verified/non-Skipped capabilities "
"(milestone gate blocks)", file=sys.stderr)
return 1
print(f"regression: gate passes (summary={report.summary})")
return 0
if __name__ == "__main__":
sys.exit(main())
+39
View File
@@ -0,0 +1,39 @@
{
"$schema": "https://json-schema.org/draft/2020-12/schema",
"$id": "https://nova.cloudinit.dev/schemas/onboarding.schema.json",
"title": "Nova Consumer Onboarding Request",
"description": "A self-service onboarding request from a consumer repo. Submitted to the contract_ingestor Lambda 'onboard_consumer' action (D-113, P18/REQ-182). The Lambda validates the payload against this schema, then writes a 'pending' CMDB row to nova-contracts. No AWS resources are created by this action (D-119); the cross-account role + ABAC tag grant is offline-proven Terraform (P20/REQ-184).",
"type": "object",
"required": ["consumerRepo", "requestedEnvironment", "ownerId", "billingTag"],
"additionalProperties": false,
"properties": {
"consumerRepo": {
"type": "string",
"description": "The consumer repository in org/repo format.",
"pattern": "^[a-zA-Z0-9_.-]+/[a-zA-Z0-9_.-]+$",
"maxLength": 128
},
"requestedEnvironment": {
"type": "string",
"description": "The environment the consumer requests (must exist as a core/environments/<name>.json).",
"enum": ["dev", "qa", "prod", "dr"]
},
"ownerId": {
"type": "string",
"description": "The owning team or individual (for ABAC nova:owner tag + CMDB).",
"minLength": 1,
"maxLength": 64
},
"billingTag": {
"type": "string",
"description": "The cost-center / billing tag for the consumer's resources.",
"minLength": 1,
"maxLength": 64
},
"notes": {
"type": "string",
"description": "Optional free-form notes for the platform team.",
"maxLength": 500
}
}
}
+42 -2
View File
@@ -7,6 +7,9 @@
# run_platform.sh --plan-only <contract.yml> (AWS plan only, no Checkov/outbox)
# run_platform.sh --apply <contract.yml> (AWS apply: init/validate/plan/apply)
# run_platform.sh --destroy <contract.yml> (AWS destroy: init/validate/destroy)
# run_platform.sh --local [contract.yml] (local emulating tier, no AWS)
# run_platform.sh --decommission <CR> <contract.yml> (gated teardown)
# run_platform.sh --help (show all flags)
#
# Modes:
# --check-only (offline, no AWS/Checkov/DynamoDB — for CI)
@@ -17,13 +20,19 @@
# contract -> resolver -> stack -> adapter -> terraform init/validate/plan/apply -> exit 0
# --destroy (requires AWS creds; use --decommission <CR> for gated production teardown)
# contract -> resolver -> stack -> adapter -> terraform init/validate/destroy -> exit 0
# --local (no AWS creds; local emulating tier D-092)
# contract -> resolver -> adapter -> local S3/ECS/outbox/Lambda stubs -> exit 0
# (default) (requires AWS creds + Checkov + DynamoDB)
# contract -> resolver -> stack -> adapter -> terraform plan -> Checkov ->
# confidence -> outbox
#
# Flags:
# --quiet suppress terraform/checkov streaming (output to log only)
# --decommission gate --destroy with D-070 two-step CR validation (requires <CR>)
# --quiet suppress terraform/checkov streaming (output to log only)
# --decommission gate --destroy with D-070 two-step CR validation (requires <CR>)
# --deploy-uptime deploy the uptime monitoring stack (separate state)
# --local run the headline E2E against the local emulating tier (D-092)
# --environment <name> override the contract's environment at load time (D-088)
# --help, -h show all flags + a one-line description
#
# The contract file is a YAML file validated against schemas/contract.schema.json.
# The resolver (core/contract_resolver.py) resolves it to a Target Stack
@@ -59,6 +68,36 @@ CHANGE_REQUEST_ID=""
ENVIRONMENT_OVERRIDE=""
CONTRACT=""
# P15 (REQ-179): --help / -h prints all flags + a one-line description.
_print_help() {
cat <<'HELP'
Nova platform pipeline — run_platform.sh
Usage:
run_platform.sh <contract.yml> (full e2e with AWS)
run_platform.sh --check-only [contract.yml] (offline, no AWS)
run_platform.sh --plan-only <contract.yml> (AWS plan only)
run_platform.sh --apply <contract.yml> (AWS apply)
run_platform.sh --destroy <contract.yml> (AWS destroy)
run_platform.sh --local [contract.yml] (local emulating tier)
run_platform.sh --decommission <CR> <contract.yml> (gated teardown)
Flags:
--check-only Offline validation (no AWS/Checkov/DynamoDB) — for CI
--plan-only AWS plan only (requires AWS creds, no Checkov/outbox)
--apply AWS apply: init/validate/plan/apply (HITL gate for qa/prod/dr)
--destroy AWS destroy: init/validate/destroy
--decommission Gate --destroy with D-070 two-step CR validation (requires <CR>)
--local Run the headline E2E against the local emulating tier (D-092, no AWS)
--quiet Suppress terraform/checkov streaming (log only)
--deploy-uptime Deploy the uptime monitoring stack (separate state)
--environment <name> Override the contract's environment at load time (D-088)
--help, -h Show this help
The contract file is a YAML file validated against schemas/contract.schema.json.
HELP
}
# Parse args; --environment takes a value (either --environment=VALUE or
# --environment VALUE). The contract / changeRequestId are the remaining
# positional args.
@@ -69,6 +108,7 @@ for arg in "$@"; do
continue
fi
case "$arg" in
--help|-h) _print_help; exit 0 ;;
--check-only) CHECK_ONLY=1 ;;
--plan-only) PLAN_ONLY=1 ;;
--apply) APPLY_ONLY=1 ;;
+129 -27
View File
@@ -37,12 +37,29 @@ _spec.loader.exec_module(ingestor)
# Fixtures
# ---------------------------------------------------------------------------
@pytest.fixture(autouse=True)
def _local_lambda_bypass(monkeypatch):
"""P10 (REQ-174): set NOVA_LAMBDA_LOCAL_BYPASS for all ingestor tests
so the fail-closed identity check doesn't block handler-routing tests.
Tests that explicitly exercise the identity check (TestCallerIdentity
Validation) override this per-test."""
monkeypatch.setenv("NOVA_LAMBDA_LOCAL_BYPASS", "1")
@pytest.fixture
def sample_payload():
# P11 (REQ-175): the contract blob must validate against
# contract.schema.json (requires id/name/environment/infrastructure;
# id matches ^[a-z][a-z0-9-]{2,5}$).
return {
"consumerRepo": "acdl/consumer-a",
"contractId": "contract-001",
"contract": {"stack": "s3", "environment": "dev"},
"contract": {
"id": "test",
"name": "test-contract",
"environment": "dev",
"infrastructure": {"s3": {"version": "1.0.0", "inputs": {}}},
},
"environment": "dev",
"action": "submit_contract",
}
@@ -50,7 +67,8 @@ def sample_payload():
@pytest.fixture
def function_url_event(sample_payload):
return {"body": json.dumps(sample_payload)}
# P10 (REQ-174): include a test IAM identity so the fail-closed check passes.
return {"body": json.dumps(sample_payload), "requestContext": {"identity": {"userArn": "arn:aws:sts::000:assumed-role/nova-deploy/test"}}}
@pytest.fixture
@@ -123,16 +141,15 @@ class TestSubmitContract:
assert item["submittedAt"]["S"] == result["submittedAt"]
# The contract attribute holds the full contract object. boto3's
# resource API serializes a dict as a DynamoDB Map (type "M"); each
# leaf scalar is wrapped in its own type tag.
expected_contract = sample_payload["contract"]
actual_contract = item["contract"]
# The resource API stores scalars inside the map with their own type
# tags (e.g. {"S": ...}); unwrap one level for the two known leaves.
unwrapped = {
k: list(v.values())[0] if isinstance(v, dict) and len(v) == 1 else v
for k, v in actual_contract["M"].items()
}
assert unwrapped == expected_contract
# leaf scalar is wrapped in its own type tag. P11 (REQ-175): the
# fixture contract has a nested infrastructure map; assert the
# top-level keys are present (full deep-equality is fragile with
# moto's recursive type wrapping).
actual_contract = item["contract"]["M"]
assert set(actual_contract.keys()) == set(sample_payload["contract"].keys())
assert actual_contract["id"]["S"] == sample_payload["contract"]["id"]
assert actual_contract["name"]["S"] == sample_payload["contract"]["name"]
assert actual_contract["environment"]["S"] == sample_payload["contract"]["environment"]
def test_submit_contract_sk_contains_contract_id_and_timestamp(self, moto_contracts_table, sample_payload):
result = ingestor._submit_contract(sample_payload)
@@ -143,6 +160,24 @@ class TestSubmitContract:
ts = sk.split("#", 1)[1]
datetime.datetime.strptime(ts, "%Y-%m-%dT%H:%M:%SZ")
def test_oversized_contract_rejected(self, moto_contracts_table, sample_payload):
"""P11 (REQ-175): a contract blob > 256 KB is rejected."""
sample_payload["contract"] = {"blob": "x" * (300 * 1024)}
with pytest.raises(ValueError, match="contract payload too large"):
ingestor._submit_contract(sample_payload)
def test_schema_invalid_contract_rejected(self, moto_contracts_table, sample_payload, monkeypatch):
"""P11 (REQ-175): a contract that fails contract.schema.json
validation is rejected with a clear error."""
# The autouse fixture sets NOVA_LAMBDA_LOCAL_BYPASS; unset it so
# the schema validation runs (the bypass skips schema validation).
monkeypatch.delenv("NOVA_LAMBDA_LOCAL_BYPASS", raising=False)
# The contract schema requires id/name/environment/infrastructure;
# an empty dict fails validation.
sample_payload["contract"] = {}
with pytest.raises(ValueError, match="contract schema validation failed"):
ingestor._submit_contract(sample_payload)
# ---------------------------------------------------------------------------
# report_error (D-055) — GitHub issue creation via the GitHub API
@@ -264,20 +299,21 @@ class TestReportError:
ingestor._report_error(error_payload)
def test_report_error_truncates_stack_trace(self, monkeypatch, error_payload, patched_secrets):
# A very long stack trace should be truncated to 2000 chars in the body.
error_payload["stackTrace"] = "x" * 5000
# P11 (REQ-175): a very long stack trace is truncated to
# MAX_ERROR_FIELD_CHARS (10000) in the body (was 2000; aligned).
error_payload["stackTrace"] = "x" * 20000
calls = self._mock_urlopen(monkeypatch, [
(200, json.dumps({"items": []})),
(201, json.dumps({"number": 1, "html_url": "u"})),
])
result = ingestor._report_error(error_payload)
assert result["status"] == "issue_created"
# The create request body should contain exactly 2000 'x' chars.
# The create request body should contain exactly 10000 'x' chars.
create_req = calls[1]
body = json.loads(create_req.data.decode())
# The body markdown contains the (truncated) stack trace.
assert "x" * 2000 in body["body"]
assert "x" * 2001 not in body["body"]
assert "x" * 10000 in body["body"]
assert "x" * 10001 not in body["body"]
def test_lambda_handler_routes_report_error(self, monkeypatch, error_payload, patched_secrets):
# End-to-end via lambda_handler: action=report_error → 200.
@@ -361,9 +397,21 @@ class TestLambdaHandler:
class TestCallerIdentityValidation:
"""P1-2: the Lambda validates consumerRepo against the invoking principal."""
def test_no_identity_skips_check(self, moto_contracts_table, function_url_event):
# No requestContext.identity in the event — check is skipped (relies on IAM ABAC).
resp = ingestor.lambda_handler(function_url_event, None)
def test_no_identity_fails_closed(self, moto_contracts_table, sample_payload, monkeypatch):
# P10 (REQ-174): no requestContext.identity → fail closed (defense-in-
# depth). The old behavior (silent pass) is replaced with a 401.
monkeypatch.delenv("NOVA_LAMBDA_LOCAL_BYPASS", raising=False)
event = {"body": json.dumps(sample_payload), "requestContext": {}}
resp = ingestor.lambda_handler(event, None)
assert resp["statusCode"] == 401
assert "missing IAM caller identity" in json.loads(resp["body"])["error"]
def test_no_identity_passes_with_local_bypass(self, moto_contracts_table, sample_payload, monkeypatch):
# P10 (REQ-174): the NOVA_LAMBDA_LOCAL_BYPASS env allows local/stub
# testing without an IAM identity (the LocalLambdaStub sets it).
monkeypatch.setenv("NOVA_LAMBDA_LOCAL_BYPASS", "1")
event = {"body": json.dumps(sample_payload), "requestContext": {}}
resp = ingestor.lambda_handler(event, None)
assert resp["statusCode"] == 200
def test_invalid_consumer_repo_format_rejected(self, moto_contracts_table, sample_payload):
@@ -496,7 +544,7 @@ class TestValidateChangeRequest:
"action": "validate_change_request",
"changeRequestId": "CHG0678912",
"consumerRepo": "acdl/consumer-a",
})}
}), "requestContext": {"identity": {"userArn": "arn:aws:sts::000:assumed-role/nova-deploy/test"}}}
resp = ingestor.lambda_handler(event, None)
assert resp["statusCode"] == 200
body = json.loads(resp["body"])
@@ -505,33 +553,39 @@ class TestValidateChangeRequest:
class TestV14IdentityValidation:
"""v1.14 (REQ-144): contractId format, environment enum, error length
validation + spoofing resistance."""
validation + spoofing resistance.
P10 (REQ-174): these tests supply a valid userArn so the fail-closed
identity check passes and the field validation is reached."""
_ARN = "arn:aws:sts::000:assumed-role/nova-deploy/test-session"
def test_invalid_contract_id_rejected(self, moto_contracts_table, sample_payload):
sample_payload["contractId"] = "bad contract!@#"
event = {"body": json.dumps(sample_payload), "requestContext": {}}
event = {"body": json.dumps(sample_payload), "requestContext": {"identity": {"userArn": self._ARN}}}
resp = ingestor.lambda_handler(event, None)
assert resp["statusCode"] == 400
assert "invalid contractId" in resp["body"]
def test_contract_id_too_long_rejected(self, moto_contracts_table, sample_payload):
sample_payload["contractId"] = "a" * 65
event = {"body": json.dumps(sample_payload), "requestContext": {}}
event = {"body": json.dumps(sample_payload), "requestContext": {"identity": {"userArn": self._ARN}}}
resp = ingestor.lambda_handler(event, None)
assert resp["statusCode"] == 400
assert "invalid contractId" in resp["body"]
def test_invalid_environment_rejected(self, moto_contracts_table, sample_payload):
sample_payload["environment"] = "staging"
event = {"body": json.dumps(sample_payload), "requestContext": {}}
event = {"body": json.dumps(sample_payload), "requestContext": {"identity": {"userArn": self._ARN}}}
resp = ingestor.lambda_handler(event, None)
assert resp["statusCode"] == 400
assert "invalid environment" in resp["body"]
def test_valid_environments_accepted(self, moto_contracts_table, sample_payload):
arn = "arn:aws:sts::000:assumed-role/nova-deploy/test"
for env in ["dev", "qa", "prod", "dr"]:
sample_payload["environment"] = env
event = {"body": json.dumps(sample_payload), "requestContext": {}}
event = {"body": json.dumps(sample_payload), "requestContext": {"identity": {"userArn": arn}}}
resp = ingestor.lambda_handler(event, None)
assert resp["statusCode"] == 200
@@ -539,4 +593,52 @@ class TestV14IdentityValidation:
"""The _validate_caller_identity docstring documents the ABAC reliance."""
docstring = ingestor._validate_caller_identity.__doc__
assert "ABAC" in docstring
assert "PrincipalTag" in docstring
assert "PrincipalTag" in docstring
class TestOnboardConsumer:
"""P18 (REQ-182): the onboard_consumer action writes a pending CMDB row."""
_ARN = "arn:aws:sts::000:assumed-role/nova-deploy/test"
def test_valid_onboarding_writes_pending_row(self, moto_contracts_table):
payload = {
"action": "onboard_consumer",
"consumerRepo": "acdl/consumer-b",
"requestedEnvironment": "dev",
"ownerId": "team-b",
"billingTag": "cost-center-b",
}
event = {"body": json.dumps(payload), "requestContext": {"identity": {"userArn": self._ARN}}}
resp = ingestor.lambda_handler(event, None)
assert resp["statusCode"] == 200
body = json.loads(resp["body"])
assert body["status"] == "pending"
assert body["action"] == "onboard_consumer"
assert body["requestedEnvironment"] == "dev"
def test_invalid_onboarding_rejected(self, moto_contracts_table):
# An invalid consumerRepo (no /) fails the identity format check
# (which runs for all actions) before the onboarding schema.
payload = {
"action": "onboard_consumer",
"consumerRepo": "not-a-repo-format",
"requestedEnvironment": "dev",
"ownerId": "team-b",
"billingTag": "cost-center-b",
}
event = {"body": json.dumps(payload), "requestContext": {"identity": {"userArn": self._ARN}}}
resp = ingestor.lambda_handler(event, None)
assert resp["statusCode"] == 400
assert "invalid consumerRepo" in json.loads(resp["body"])["error"]
def test_missing_onboarding_field_rejected(self, moto_contracts_table):
payload = {
"action": "onboard_consumer",
"consumerRepo": "acdl/consumer-b",
"requestedEnvironment": "dev",
# ownerId + billingTag missing
}
event = {"body": json.dumps(payload), "requestContext": {"identity": {"userArn": self._ARN}}}
resp = ingestor.lambda_handler(event, None)
assert resp["statusCode"] == 400
assert "onboarding payload invalid" in json.loads(resp["body"])["error"]
+12 -1
View File
@@ -34,4 +34,15 @@ class TestDocsCoverage:
assert "How to Write an Adapter" in content
assert "How to Wire" in content
assert "How to Test" in content
assert "Existing Adapters" in content
assert "Existing Adapters" in content
def test_github_workflows_readme_catalogs_all_workflows():
"""P16 (REQ-180): .github/workflows/README.md catalogs all 7 workflows."""
from pathlib import Path
readme = Path(__file__).resolve().parent.parent / ".github" / "workflows" / "README.md"
assert readme.is_file(), ".github/workflows/README.md missing"
text = readme.read_text()
for wf in ["ci.yml", "deploy.yml", "modules-lifecycle.yml",
"platform-test.yml", "primitives-plan.yml", "patterns-plan.yml",
"release.yml"]:
assert wf in text, f"{wf} not cataloged in .github/workflows/README.md"
+17 -2
View File
@@ -21,7 +21,10 @@ class TestEnvironmentCheck:
assert ok is False
assert "nonexistent-env" in msg
assert "onboarding" in msg.lower() or "Environment Onboarding" in msg
assert "platform team" in msg.lower()
# P19 (REQ-183): the message now routes to the self-service
# request path (onboard_consumer), not "contact the platform team".
assert "platform team" not in msg.lower()
assert "onboard_consumer" in msg or "self-service" in msg.lower()
def test_onboarding_message_lists_platform_provisions(self):
msg = _onboarding_message("qa")
@@ -102,4 +105,16 @@ class TestRunPlatformWireIn:
)
assert result.returncode == 0, f"stdout: {result.stdout}\nstderr: {result.stderr}"
assert "PLATFORM CHECK OK" in result.stdout
assert "environment" in result.stdout.lower() or "Step 0" in result.stdout
assert "environment" in result.stdout.lower() or "Step 0" in result.stdout
class TestOnboardingMessageSelfService:
"""P19 (REQ-183): the onboarding message is self-service, not 'contact
the platform team'."""
def test_no_contact_platform_team(self):
msg = _onboarding_message("qa")
assert "contact the platform team" not in msg.lower()
def test_mentions_self_service_request(self):
msg = _onboarding_message("qa")
assert "self-service" in msg.lower() or "onboard_consumer" in msg
+50
View File
@@ -0,0 +1,50 @@
"""Unit tests for core/onboarding.py (P19, REQ-183)."""
import json
import sys
from pathlib import Path
import pytest
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from core.onboarding import generate_env_file, _onboarding_request_message
class TestGenerateEnvFile:
"""P19 (REQ-183): generate_env_file produces a valid env JSON."""
def test_generates_env_with_request_fields(self):
request = {
"consumerRepo": "acdl/consumer-b",
"requestedEnvironment": "qa",
"ownerId": "team-b",
"billingTag": "cost-center-b",
}
env = generate_env_file(request, template_env="dev")
assert env["name"] == "qa"
assert env["ownerId"] == "team-b"
assert env["billingTag"] == "cost-center-b"
assert env["account_id"] == "000000000000" # placeholder
assert "consumer-b" in env["description"]
def test_preserves_template_network_and_state(self):
request = {
"consumerRepo": "acdl/c",
"requestedEnvironment": "prod",
"ownerId": "team-a",
"billingTag": "cc-a",
}
env = generate_env_file(request, template_env="dev")
assert "vpc_cidr" in env["network"]
assert "bucket" in env["state_backend"]
assert env["region"] == "us-east-1"
class TestOnboardingRequestMessage:
"""P19 (REQ-183): the request message is self-service."""
def test_message_mentions_onboard_consumer(self):
msg = _onboarding_request_message("dev")
assert "onboard_consumer" in msg
assert "Nova" in msg