From 01494b2082867daf5e0b028fba5af80a04170818 Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 21 Jun 2026 08:44:49 +0000 Subject: [PATCH] fix: eight correctness bugs in deterministic helpers + regression tests (v0.6.6) A correctness audit of the helpers (the audit that did not finish in the v0.6.3 round) found eight real bugs, each with a reproduction; all fixed and pinned with a regression test (tests/test_v066_bugfixes.py). The missing tests are why they shipped green. No new features. - cross_account_access_validator: a Deny expressed via NotAction/NotResource/ NotPrincipal was silently dropped -> false "allow" from a security-decision tool. Now evaluates the inverted forms and flags an open-question caveat. - throttle_tuning_recommender.parse_size: decimal units parsed as binary (64MB -> 67108864). Now honours the 'i' (MiB=1024, MB=1000), matching the sibling helper. - small_object_analyzer: "GLACIER IR"/"Glacier Instant Retrieval" misclassified as plain GLACIER (40KB vs 128KB min-billable). Now normalises class spelling. - migration_cost_estimator: CSV total_size_bytes="0" shadowed the GB/TB fallback (presence check now); malformed stdin/missing file now emit the {ok:false} envelope instead of a traceback. - notification_target_policy_validator: a wildcard Principal:"*" Allow was reported as "no S3 grant" (false "delivery blocked"); now permits delivery + flags over-broad. - lifecycle_rule_simulator: same-day transition+expiration now sorts with expiration precedence (S3 semantics), avoiding false "wasted days". - parse_sigv4_error: no longer mislabels a short canonical-request fragment's last line as payload_hash (requires >= 6 lines). Validation: all gates green; 219 pytest (+8) + 21 extension tests; 31/31 baselines PASS. Version 0.6.5 -> 0.6.6. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01QhkyBLeDtBTmvTdVBqgCAV --- CHANGELOG.md | 34 ++++++ docs/ARCHITECTURE.md | 2 +- docs/cli-reference.md | 2 +- pyproject.toml | 2 +- skill-registry.yaml | 2 +- .../notification_target_policy_validator.py | 23 +++- .../scripts/lifecycle_rule_simulator.py | 5 +- .../scripts/small_object_analyzer.py | 8 +- .../scripts/migration_cost_estimator.py | 43 +++++--- .../scripts/throttle_tuning_recommender.py | 8 +- .../scripts/parse_sigv4_error.py | 5 +- .../scripts/cross_account_access_validator.py | 80 +++++++++----- tests/test_v066_bugfixes.py | 103 ++++++++++++++++++ 13 files changed, 265 insertions(+), 52 deletions(-) create mode 100644 tests/test_v066_bugfixes.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 0876b4c..75a0899 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,39 @@ # Changelog +## 2026-06-21 — v0.6.6: Correctness hardening (helper bug-fix release) + +A focused bug-fix release. A correctness audit of the deterministic helpers (the +audit that did not complete in the v0.6.3 round) found eight real bugs, each with a +reproduction; all are fixed and pinned with a regression test (the missing tests are +why they shipped green). No new features; no new complexity. + +- **`cross_account_access_validator` silently dropped a Deny expressed via + `NotAction`/`NotResource`/`NotPrincipal`** → false "allow" from a security-decision + tool (highest impact). Now evaluates the inverted forms correctly (Deny-NotAction + blocks the non-excluded action; the excluded action stays allowed) and surfaces an + open-question caveat for inverted clauses. +- **`throttle_tuning_recommender.parse_size` parsed decimal units as binary** + (`64MB`→67108864 instead of 64000000), contradicting the sibling + `multipart_etag_calculator`. Fixed to honour the `i` (MiB=1024, MB=1000). +- **`small_object_analyzer` misclassified "GLACIER IR" / "Glacier Instant Retrieval" + as plain GLACIER** (40 KB min-billable instead of 128 KB), under-reporting the + small-object penalty. Now normalises class spelling (space/hyphen/prose). +- **`migration_cost_estimator`**: a CSV `total_size_bytes="0"` (truthy string) + shadowed the `total_size_gb`/`tb` fallback → 0-byte migration; now a presence/value + check. Malformed stdin JSON / missing CSV now emit the `{"ok":false}` envelope + instead of a traceback. +- **`notification_target_policy_validator`**: a wildcard `Principal:"*"` Allow was + reported as "no S3 grant" (false "delivery blocked"); it now correctly permits + delivery and flags the grant as over-broad. +- **`lifecycle_rule_simulator`**: same-day transition+expiration now sorts + deterministically with expiration precedence (S3 semantics), avoiding a false + "wasted days" on a class never really entered. +- **`parse_sigv4_error`**: stopped mislabeling a short canonical-request fragment's + last line as `payload_hash` (requires a full ≥6-line canonical request). + +Validation: all gates green; 219 pytest (+8 regression tests) + 21 extension tests; +31/31 baselines 100% PASS. Version 0.6.5 → 0.6.6. + ## 2026-06-21 — v0.6.5: Thicken non-AWS provider depth Continues the multi-provider rebalancing (v0.6.4) by deepening the actual content a diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index f6abe5f..f6bd34b 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -1,6 +1,6 @@ # Architecture -StorageOps v0.6.5 is a Pi Coding Agent extension and skill pack. +StorageOps v0.6.6 is a Pi Coding Agent extension and skill pack. ## Components diff --git a/docs/cli-reference.md b/docs/cli-reference.md index 3693f83..ce5ab53 100644 --- a/docs/cli-reference.md +++ b/docs/cli-reference.md @@ -41,7 +41,7 @@ from the older local package. Deployment provenance is written to ## `storageops --version` ```text -StorageOps v0.6.5 (pi: 0.78.0) +StorageOps v0.6.6 (pi: 0.78.0) httpmon : /root/.storageops/bin/httpmon api key : api-key file independent install : yes (~/.storageops/agent) diff --git a/pyproject.toml b/pyproject.toml index e56ad6b..298199c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "storageops" -version = "0.6.5" +version = "0.6.6" description = "StorageOps — S3-compatible object storage diagnostic toolkit. A Pi Coding Agent extension + skill pack." requires-python = ">=3.11" readme = "README.md" diff --git a/skill-registry.yaml b/skill-registry.yaml index 12a6ea1..5f6188c 100644 --- a/skill-registry.yaml +++ b/skill-registry.yaml @@ -1,4 +1,4 @@ -# StorageOps Skill Registry v0.6.5 +# StorageOps Skill Registry v0.6.6 # Machine-readable metadata for all StorageOps skills. # Skills are auto-discovered by Pi from the skills/ directory. diff --git a/skills/storageops-event-notification/scripts/notification_target_policy_validator.py b/skills/storageops-event-notification/scripts/notification_target_policy_validator.py index 39c2fbf..02f26f1 100644 --- a/skills/storageops-event-notification/scripts/notification_target_policy_validator.py +++ b/skills/storageops-event-notification/scripts/notification_target_policy_validator.py @@ -64,10 +64,22 @@ def _statements(policy: dict) -> list: return [s for s in _as_list(policy.get("Statement")) if isinstance(s, dict)] -def _principal_is_s3(stmt: dict) -> bool: +def _principal_is_wildcard(stmt: dict) -> bool: principal = stmt.get("Principal") if principal == "*": - return False + return True + if isinstance(principal, dict): + return "*" in _as_list(principal.get("AWS")) or "*" in _as_list(principal.get("Service")) + return False + + +def _principal_is_s3(stmt: dict) -> bool: + # A wildcard principal ("*", {"AWS":"*"}) DOES allow s3.amazonaws.com to invoke + # the target — over-broad, but it permits delivery, so it must not be reported + # as "no S3 grant" (false "delivery blocked"). + if _principal_is_wildcard(stmt): + return True + principal = stmt.get("Principal") if isinstance(principal, str): return principal == S3_SERVICE if isinstance(principal, dict): @@ -190,7 +202,13 @@ def validate(policy: dict, target_type: Optional[str], bucket_arn: Optional[str] policy_ok = not missing + overbroad = policy_ok and any(_principal_is_wildcard(s) for s in action_stmts) if policy_ok: + wildcard_note = ( + " Note: the grant uses a wildcard Principal (\"*\"), which permits delivery but is " + "over-broad — scope it to Principal Service s3.amazonaws.com with an aws:SourceArn " + "condition." if overbroad else "" + ) summary = ( f"Target resource policy permits S3 delivery: an Allow statement grants " f"{action} to Principal {S3_SERVICE}" @@ -200,6 +218,7 @@ def validate(policy: dict, target_type: Optional[str], bucket_arn: Optional[str] "Policy is correct. If events still are not delivered, the target policy is not " "the cause — check the bucket notification rule (event type + prefix/suffix filter) " "with notification_config_analyzer.py, and confirm CloudTrail shows the event was emitted." + + wildcard_note ) elif sourcearn_mismatch: summary = ( diff --git a/skills/storageops-lifecycle-cost/scripts/lifecycle_rule_simulator.py b/skills/storageops-lifecycle-cost/scripts/lifecycle_rule_simulator.py index 555ed29..7085f3f 100644 --- a/skills/storageops-lifecycle-cost/scripts/lifecycle_rule_simulator.py +++ b/skills/storageops-lifecycle-cost/scripts/lifecycle_rule_simulator.py @@ -203,7 +203,10 @@ def analyze(rules: list, age_days, avg_size, object_count, start_class: str) -> "orphaned/incomplete multipart parts remain billable" ) - events.sort(key=lambda e: e[0]) + # Deterministic tiebreak: when a transition and an expiration land on the same + # day, S3 gives expiration precedence (the transition does not happen), so sort + # expire before transition rather than relying on insertion order. + events.sort(key=lambda e: (e[0], 0 if e[1] == "expire" else 1)) # Walk the timeline to determine class residency windows. A class entered at # day D and left at day L bills for max(L-D, min_duration_days). If the diff --git a/skills/storageops-lifecycle-cost/scripts/small_object_analyzer.py b/skills/storageops-lifecycle-cost/scripts/small_object_analyzer.py index 52710b6..8a870c5 100644 --- a/skills/storageops-lifecycle-cost/scripts/small_object_analyzer.py +++ b/skills/storageops-lifecycle-cost/scripts/small_object_analyzer.py @@ -5,7 +5,7 @@ CSV columns (required header): key,size_bytes,storage_class Output: JSON {ok, summary, class_breakdown, recommendations, details}""" -import argparse, csv, json, sys +import argparse, csv, json, re, sys # Minimum billable object size (bytes) per storage class family _MIN_BILLABLE = { @@ -23,7 +23,11 @@ def _min_billable(storage_class: str) -> int: - upper = storage_class.upper() + # Normalise spelling so "GLACIER IR" / "Glacier Instant Retrieval" / "glacier-ir" + # all match GLACIER_IR rather than falling through to plain GLACIER. + upper = re.sub(r"[ \-]+", "_", storage_class.upper()) + if "INSTANT" in upper and "GLACIER" in upper: + upper = "GLACIER_IR" for marker in sorted(_MIN_BILLABLE, key=len, reverse=True): if marker in upper: return _MIN_BILLABLE[marker] diff --git a/skills/storageops-migration-sync/scripts/migration_cost_estimator.py b/skills/storageops-migration-sync/scripts/migration_cost_estimator.py index b7efbee..7de1fb7 100755 --- a/skills/storageops-migration-sync/scripts/migration_cost_estimator.py +++ b/skills/storageops-migration-sync/scripts/migration_cost_estimator.py @@ -37,14 +37,27 @@ } +def _present(row: dict, key: str) -> bool: + # CSV cells arrive as strings, so a literal "0" is truthy; test presence and a + # non-empty value, not truthiness, otherwise total_size_bytes="0" wrongly + # shadows a total_size_gb fallback. + return str(row.get(key, "")).strip() != "" + + def _resolve_total_bytes(row: dict) -> float: - """Byte-priority: total_size_bytes > total_size_gb > total_size_tb.""" - if row.get("total_size_bytes"): + """Byte-priority: total_size_bytes > total_size_gb > total_size_tb. + + A present-but-zero byte value is only used when it is the sole size given; + otherwise fall through to the GB/TB columns. + """ + if _present(row, "total_size_bytes") and float(row["total_size_bytes"]) > 0: return float(row["total_size_bytes"]) - if row.get("total_size_gb"): + if _present(row, "total_size_gb"): return float(row["total_size_gb"]) * 1e9 - if row.get("total_size_tb"): + if _present(row, "total_size_tb"): return float(row["total_size_tb"]) * 1e12 + if _present(row, "total_size_bytes"): + return float(row["total_size_bytes"]) # explicit 0 with no other size given raise ValueError("missing one of: total_size_bytes, total_size_gb, total_size_tb") @@ -168,18 +181,22 @@ def main() -> None: ap.add_argument("--pretty", action="store_true", help="Pretty-print JSON output") args = ap.parse_args() - inputs: list[dict] = [] - if args.stdin: - inputs = _parse_stdin() - elif args.file: - inputs = _parse_csv(args.file) - else: - ap.print_help() - raise SystemExit("specify --file or --stdin") + indent = 2 if args.pretty else None + try: + if args.stdin: + inputs = _parse_stdin() + elif args.file: + inputs = _parse_csv(args.file) + else: + ap.print_help() + raise SystemExit("specify --file or --stdin") + except (json.JSONDecodeError, OSError) as exc: + # Bad/unreadable input must yield the JSON error envelope, not a traceback. + print(json.dumps({"ok": False, "error": f"could not read input: {exc}"}, indent=indent)) + return results = [estimate(row) for row in inputs] out = results[0] if len(results) == 1 else results - indent = 2 if args.pretty else None print(json.dumps(out, indent=indent)) diff --git a/skills/storageops-performance-diagnosis/scripts/throttle_tuning_recommender.py b/skills/storageops-performance-diagnosis/scripts/throttle_tuning_recommender.py index 56d149c..1e6c363 100644 --- a/skills/storageops-performance-diagnosis/scripts/throttle_tuning_recommender.py +++ b/skills/storageops-performance-diagnosis/scripts/throttle_tuning_recommender.py @@ -72,11 +72,11 @@ def parse_size(value: Optional[str]) -> Optional[int]: raise ValueError(f"unparseable size: {value!r}") num = float(m.group(1)) unit = m.group(2).lower() - binary = bool(m.group(3)) - factor = {"": 1, "k": 1, "m": 2, "g": 3, "t": 4}[unit] - base = 1024 if binary or unit else 1000 + binary = bool(m.group(3)) # the 'i' in MiB/GiB selects 1024, not the unit letter if unit == "": - base = 1 + return int(num) + factor = {"k": 1, "m": 2, "g": 3, "t": 4}[unit] + base = 1024 if binary else 1000 return int(num * (base ** factor)) diff --git a/skills/storageops-s3-protocol-compatibility/scripts/parse_sigv4_error.py b/skills/storageops-s3-protocol-compatibility/scripts/parse_sigv4_error.py index a1f9354..dcdb65b 100755 --- a/skills/storageops-s3-protocol-compatibility/scripts/parse_sigv4_error.py +++ b/skills/storageops-s3-protocol-compatibility/scripts/parse_sigv4_error.py @@ -90,7 +90,10 @@ def _canonical_summary(canonical_request: str) -> dict[str, Any]: summary["path"] = lines[1] if len(lines) > 2: summary["query"] = lines[2] - if len(lines) >= 2: + # The payload hash is the last line of a *complete* canonical request (method, + # path, query, headers, blank, signed-headers, hash = >= 6 lines). On a short + # fragment lines[-1] is the path/query, not the hash — don't mislabel it. + if len(lines) >= 6: summary["payload_hash"] = lines[-1] signed_headers_index = None for index, line in enumerate(lines): diff --git a/skills/storageops-security-iam-policy/scripts/cross_account_access_validator.py b/skills/storageops-security-iam-policy/scripts/cross_account_access_validator.py index c82a42c..64c1eba 100644 --- a/skills/storageops-security-iam-policy/scripts/cross_account_access_validator.py +++ b/skills/storageops-security-iam-policy/scripts/cross_account_access_validator.py @@ -73,24 +73,28 @@ def _glob_match(pattern: str, value: str) -> bool: return re.match(regex, value) is not None -def _action_matches(stmt_actions: Any, action: str) -> bool: - for a in _as_list(stmt_actions): - if _glob_match(str(a), action): - return True - return False - - -def _resource_matches(stmt_resources: Any, resource: str) -> bool: - resources = _as_list(stmt_resources) - if not resources: - # Identity policies always carry Resource; resource policies may omit it - # (meaning "this resource"). Treat an absent Resource as a match. - return True - return any(_glob_match(str(r), resource) for r in resources) - - -def _principal_matches(stmt_principal: Any, principal_arn: str) -> bool: - """Does a resource-policy Principal grant the given caller principal ARN?""" +def _action_matches(stmt: dict, action: str) -> tuple[bool, bool]: + """(matches, used_not). Honour Action and the inverted NotAction form.""" + if "Action" in stmt: + return any(_glob_match(str(a), action) for a in _as_list(stmt["Action"])), False + if "NotAction" in stmt: + return (not any(_glob_match(str(a), action) for a in _as_list(stmt["NotAction"]))), True + return True, False # no action constraint + + +def _resource_matches(stmt: dict, resource: str) -> tuple[bool, bool]: + """(matches, used_not). Honour Resource and the inverted NotResource form.""" + if "Resource" in stmt: + return any(_glob_match(str(r), resource) for r in _as_list(stmt["Resource"])), False + if "NotResource" in stmt: + return (not any(_glob_match(str(r), resource) for r in _as_list(stmt["NotResource"]))), True + # Identity policies always carry Resource; resource policies may omit it + # (meaning "this resource"). Treat an absent Resource as a match. + return True, False + + +def _principal_grants(stmt_principal: Any, principal_arn: str) -> bool: + """Does a resource-policy Principal value grant the given caller principal ARN?""" if stmt_principal == "*": return True values: List[str] = [] @@ -115,6 +119,16 @@ def _principal_matches(stmt_principal: Any, principal_arn: str) -> bool: return False +def _principal_matches(stmt: dict, principal_arn: str) -> tuple[bool, bool]: + """(matches, used_not). Honour Principal and the inverted NotPrincipal form.""" + if "Principal" in stmt: + return _principal_grants(stmt["Principal"], principal_arn), False + if "NotPrincipal" in stmt: + # A statement applies to every principal NOT listed in NotPrincipal. + return (not _principal_grants(stmt["NotPrincipal"], principal_arn)), True + return False, False # resource-policy statement with no principal grants nobody + + def _has_condition(stmt: dict) -> bool: return isinstance(stmt.get("Condition"), dict) and bool(stmt.get("Condition")) @@ -129,23 +143,33 @@ def evaluate_link( """Evaluate one policy: explicit Deny > explicit Allow > implicit Deny.""" allow_hit: Optional[str] = None conditional = False + allow_not = False for i, stmt in enumerate(statements, start=1): effect = str(stmt.get("Effect", "")).lower() - if not _action_matches(stmt.get("Action"), action): + a_match, a_not = _action_matches(stmt, action) + if not a_match: continue - if not _resource_matches(stmt.get("Resource"), resource): - continue - if match_principal and not _principal_matches(stmt.get("Principal"), principal_arn or ""): + r_match, r_not = _resource_matches(stmt, resource) + if not r_match: continue + used_not = a_not or r_not + if match_principal: + p_match, p_not = _principal_matches(stmt, principal_arn or "") + if not p_match: + continue + used_not = used_not or p_not sid = stmt.get("Sid", f"statement#{i}") if effect == "deny": - return {"result": "explicit_deny", "matched_sid": sid, "conditional": _has_condition(stmt)} + return {"result": "explicit_deny", "matched_sid": sid, + "conditional": _has_condition(stmt), "not_clause": used_not} if effect == "allow" and allow_hit is None: allow_hit = sid conditional = _has_condition(stmt) + allow_not = used_not if allow_hit is not None: - return {"result": "allow", "matched_sid": allow_hit, "conditional": conditional} - return {"result": "implicit_deny", "matched_sid": None, "conditional": False} + return {"result": "allow", "matched_sid": allow_hit, "conditional": conditional, + "not_clause": allow_not} + return {"result": "implicit_deny", "matched_sid": None, "conditional": False, "not_clause": False} def _kms_action_for(action: str) -> Optional[str]: @@ -283,6 +307,12 @@ def validate( "One or more matched statements carry a Condition; this validator matches the " "statement but does not evaluate Condition values — confirm the request meets them." ) + if any(l.get("not_clause") for l in links): + open_questions.append( + "A matched statement uses an inverted element (NotAction/NotResource/NotPrincipal); " + "these are evaluated as 'everything except the listed set' and are easy to get wrong — " + "double-check the intended effect." + ) if cross_account and bucket_policy is not None and identity_policy is None: open_questions.append("Caller IAM policy was not supplied; the identity-side allow is assumed unknown.") diff --git a/tests/test_v066_bugfixes.py b/tests/test_v066_bugfixes.py new file mode 100644 index 0000000..a61e076 --- /dev/null +++ b/tests/test_v066_bugfixes.py @@ -0,0 +1,103 @@ +"""Regression tests for the v0.6.6 correctness fixes in the deterministic helpers. + +Each test pins a bug found in the v0.6.6 bug-hunt so it cannot silently return. +""" + +from __future__ import annotations + +import importlib.util +import json +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] + + +def _load(rel: str, name: str): + p = ROOT / rel + spec = importlib.util.spec_from_file_location(name, p) + assert spec and spec.loader + m = importlib.util.module_from_spec(spec) + spec.loader.exec_module(m) + return m + + +# --- #1 throttle_tuning_recommender.parse_size: decimal vs binary units -------- +def test_throttle_parse_size_decimal_vs_binary(): + t = _load("skills/storageops-performance-diagnosis/scripts/throttle_tuning_recommender.py", "thr") + assert t.parse_size("64MB") == 64_000_000 + assert t.parse_size("64MiB") == 67_108_864 + assert t.parse_size("2GB") == 2_000_000_000 + assert t.parse_size("1KB") == 1000 + assert t.parse_size("1KiB") == 1024 + assert t.parse_size("1048576") == 1_048_576 # raw bytes unchanged + + +# --- #2 small_object_analyzer._min_billable: GLACIER IR spelling variants ------ +def test_small_object_glacier_ir_spellings(): + s = _load("skills/storageops-lifecycle-cost/scripts/small_object_analyzer.py", "soa") + assert s._min_billable("GLACIER_IR") == 128 << 10 + assert s._min_billable("GLACIER IR") == 128 << 10 + assert s._min_billable("Glacier Instant Retrieval") == 128 << 10 + assert s._min_billable("glacier-ir") == 128 << 10 + assert s._min_billable("GLACIER") == 40 << 10 # plain Glacier unchanged + + +# --- #3 cross_account_access_validator: Deny via NotAction must not be dropped -- +def test_cross_account_deny_notaction_blocks(): + m = _load("skills/storageops-security-iam-policy/scripts/cross_account_access_validator.py", "cav") + idp = {"Statement": [ + {"Effect": "Allow", "Action": "s3:*", "Resource": "*"}, + {"Effect": "Deny", "NotAction": "s3:GetObject", "Resource": "*"}, + ]} + r = m.validate("arn:aws:iam::111:user/a", "s3:PutObject", "arn:aws:s3:::b/k", idp, None) + idlink = next(l for l in r["links"] if l["link"] == "identity_policy") + assert idlink["result"] == "explicit_deny" + assert any("inverted" in q for q in r["open_questions"]) + # The excluded action is still allowed. + r2 = m.validate("arn:aws:iam::111:user/a", "s3:GetObject", "arn:aws:s3:::b/k", idp, None) + assert next(l for l in r2["links"] if l["link"] == "identity_policy")["result"] == "allow" + + +# --- #4 migration_cost_estimator: total_size_bytes="0" must fall back to GB ----- +def test_migration_zero_bytes_falls_back(tmp_path): + m = _load("skills/storageops-migration-sync/scripts/migration_cost_estimator.py", "mce") + row = {"object_count": "10", "total_size_bytes": "0", "total_size_gb": "500", "bandwidth_mbps": "1000"} + assert m._resolve_total_bytes(row) == 500 * 1e9 + # explicit 0 with no other size still resolves to 0, not an error + assert m._resolve_total_bytes({"total_size_bytes": "0"}) == 0 + + +# --- #7 migration_cost_estimator: bad I/O yields the JSON error envelope -------- +def test_migration_bad_input_envelope(capsys, monkeypatch): + m = _load("skills/storageops-migration-sync/scripts/migration_cost_estimator.py", "mce2") + monkeypatch.setattr("sys.argv", ["x", "--file", "/no/such/file.csv"]) + m.main() + out = json.loads(capsys.readouterr().out) + assert out["ok"] is False and "error" in out + + +# --- #5 notification_target_policy_validator: wildcard principal permits delivery +def test_notification_wildcard_principal_permits(): + m = _load("skills/storageops-event-notification/scripts/notification_target_policy_validator.py", "ntp") + policy = {"Statement": [{"Effect": "Allow", "Principal": "*", "Action": "lambda:InvokeFunction"}]} + r = m.validate(policy, "lambda", None) + assert r["policy_ok"] is True + assert "over-broad" in r["recommendation"] + + +# --- #8 lifecycle_rule_simulator: same-day expire precedes transition ----------- +def test_lifecycle_same_day_expire_precedence(): + m = _load("skills/storageops-lifecycle-cost/scripts/lifecycle_rule_simulator.py", "lrs") + events = [(30, "transition", "GLACIER"), (30, "expire", None), (10, "transition", "STANDARD_IA")] + events.sort(key=lambda e: (e[0], 0 if e[1] == "expire" else 1)) + same_day = [e for e in events if e[0] == 30] + assert same_day[0][1] == "expire" # expire sorts first on the tie + + +# --- #6 parse_sigv4_error: short fragment must not mislabel payload_hash -------- +def test_sigv4_short_fragment_no_payload_hash(): + m = _load("skills/storageops-s3-protocol-compatibility/scripts/parse_sigv4_error.py", "psv") + short = m._canonical_summary("GET\n/path\n") + assert "payload_hash" not in short + full = m._canonical_summary("GET\n/path\n\nhost;x-amz-date\n\nhost;x-amz-date\nABCDEF") + assert full.get("payload_hash") == "ABCDEF"