From b9ef64bc9eaa54a07e24ffe488b405fa616fb328 Mon Sep 17 00:00:00 2001 From: Crt Ahlin Date: Wed, 25 Feb 2026 21:15:40 +0100 Subject: [PATCH 1/2] Add remaining examples: batch processing, stamp management, CI/CD, verification (#23, #26, #27, #28) Complete the examples epic (#18) with the final four examples: - 04-batch-processing: stamp reuse across multiple uploads with --stamp-id - 07-stamp-management: full stamp lifecycle (list, info, check, pool-status) - 08-ci-cd-integration: artifact archival with GitHub Actions/GitLab CI configs - 09-verification: tamper detection, SHA-256 integrity, --verify flag --- examples/04-batch-processing/README.md | 96 ++ examples/04-batch-processing/batch_upload.py | 178 +++ examples/04-batch-processing/demo.sh | 159 +++ examples/04-batch-processing/run_demo.py | 216 +++ .../sample_files/log_entry_001.json | 18 + .../sample_files/log_entry_002.json | 18 + .../sample_files/log_entry_003.json | 19 + examples/07-stamp-management/README.md | 105 ++ examples/07-stamp-management/demo.sh | 96 ++ examples/07-stamp-management/run_demo.py | 148 ++ examples/07-stamp-management/sample_data.txt | 20 + .../07-stamp-management/stamp_lifecycle.py | 167 +++ examples/08-ci-cd-integration/README.md | 101 ++ .../08-ci-cd-integration/archive_artifacts.py | 148 ++ examples/08-ci-cd-integration/demo.sh | 135 ++ .../08-ci-cd-integration/github-action.yml | 58 + examples/08-ci-cd-integration/gitlab-ci.yml | 42 + examples/08-ci-cd-integration/run_demo.py | 177 +++ .../sample_artifacts/build_info.json | 26 + .../sample_artifacts/release_notes.txt | 26 + examples/09-verification/README.md | 111 ++ examples/09-verification/demo.sh | 130 ++ examples/09-verification/integrity_checker.py | 133 ++ examples/09-verification/run_demo.py | 185 +++ examples/09-verification/sample_document.txt | 35 + .../sample_document_tampered.txt | 35 + examples/09-verification/tamper_detection.py | 108 ++ examples/README.md | 12 + tests/test_examples.py | 1222 +++++++++++++++++ 29 files changed, 3924 insertions(+) create mode 100644 examples/04-batch-processing/README.md create mode 100644 examples/04-batch-processing/batch_upload.py create mode 100755 examples/04-batch-processing/demo.sh create mode 100644 examples/04-batch-processing/run_demo.py create mode 100644 examples/04-batch-processing/sample_files/log_entry_001.json create mode 100644 examples/04-batch-processing/sample_files/log_entry_002.json create mode 100644 examples/04-batch-processing/sample_files/log_entry_003.json create mode 100644 examples/07-stamp-management/README.md create mode 100755 examples/07-stamp-management/demo.sh create mode 100644 examples/07-stamp-management/run_demo.py create mode 100644 examples/07-stamp-management/sample_data.txt create mode 100644 examples/07-stamp-management/stamp_lifecycle.py create mode 100644 examples/08-ci-cd-integration/README.md create mode 100644 examples/08-ci-cd-integration/archive_artifacts.py create mode 100755 examples/08-ci-cd-integration/demo.sh create mode 100644 examples/08-ci-cd-integration/github-action.yml create mode 100644 examples/08-ci-cd-integration/gitlab-ci.yml create mode 100644 examples/08-ci-cd-integration/run_demo.py create mode 100644 examples/08-ci-cd-integration/sample_artifacts/build_info.json create mode 100644 examples/08-ci-cd-integration/sample_artifacts/release_notes.txt create mode 100644 examples/09-verification/README.md create mode 100755 examples/09-verification/demo.sh create mode 100644 examples/09-verification/integrity_checker.py create mode 100644 examples/09-verification/run_demo.py create mode 100644 examples/09-verification/sample_document.txt create mode 100644 examples/09-verification/sample_document_tampered.txt create mode 100644 examples/09-verification/tamper_detection.py diff --git a/examples/04-batch-processing/README.md b/examples/04-batch-processing/README.md new file mode 100644 index 0000000..d26069c --- /dev/null +++ b/examples/04-batch-processing/README.md @@ -0,0 +1,96 @@ +# Example 04: Batch Processing + +Demonstrates uploading multiple files efficiently by reusing a single postage stamp. + +## What This Demonstrates + +- Uploading multiple files with stamp reuse (`--stamp-id`) +- Using `--size medium` for appropriate stamp sizing +- Extracting stamp ID from verbose (`-v`) output +- Building a manifest (JSON) mapping filenames to Swarm references +- Downloading and verifying one file from the batch + +## Use Case + +When archiving multiple log files, dataset partitions, or document collections, purchasing a new stamp for each file is wasteful and slow. By capturing the stamp ID from the first upload and reusing it for subsequent uploads, you skip the stamp purchase step entirely — reducing upload time from ~60 seconds to ~5 seconds per file. + +## Prerequisites + +1. Install the CLI: `pip install -e .` +2. Gateway access (default, no setup needed) + +## Quick Start + +### Shell + +```bash +chmod +x demo.sh +./demo.sh +``` + +### Python + +```bash +python run_demo.py +``` + +## Step-by-Step Walkthrough + +### 1. Upload first file with verbose output + +The first upload uses `--size medium -v` to get detailed output including the stamp ID: + +```bash +swarm-prov-upload upload --file sample_files/log_entry_001.json --size medium -v --usePool +``` + +Extract the stamp ID from the verbose output line: `Stamp ID Received: <64-char-hex>` + +### 2. Upload remaining files with stamp reuse + +Subsequent uploads skip stamp purchase by providing the captured stamp ID: + +```bash +swarm-prov-upload upload --file sample_files/log_entry_002.json --stamp-id +swarm-prov-upload upload --file sample_files/log_entry_003.json --stamp-id +``` + +### 3. Build manifest + +Create a JSON file mapping each filename to its Swarm reference for later retrieval. + +### 4. Download and verify + +```bash +swarm-prov-upload download --output-dir ./downloads +``` + +Compare SHA-256 hashes to verify integrity. + +## Batch Upload Helper + +The `batch_upload.py` script automates the full batch workflow: + +```bash +python batch_upload.py --directory ./sample_files +python batch_upload.py --directory ./sample_files --std "PROV-STD-V1" +``` + +It handles stamp capture, reuse, and manifest generation automatically. + +## Sample Files + +| File | Description | +|------|-------------| +| `sample_files/log_entry_001.json` | INFO: User login event | +| `sample_files/log_entry_002.json` | WARNING: Rate limit approached | +| `sample_files/log_entry_003.json` | ERROR: Transaction failure | + +## Files + +| File | Description | +|------|-------------| +| `demo.sh` | Shell demo — batch upload with stamp reuse | +| `run_demo.py` | Python demo — same workflow with argparse support | +| `batch_upload.py` | Standalone batch upload tool with manifest generation | +| `sample_files/` | Directory of sample log entries | diff --git a/examples/04-batch-processing/batch_upload.py b/examples/04-batch-processing/batch_upload.py new file mode 100644 index 0000000..1355b71 --- /dev/null +++ b/examples/04-batch-processing/batch_upload.py @@ -0,0 +1,178 @@ +#!/usr/bin/env python3 +""" +Batch Upload Tool + +Uploads all files in a directory to Swarm with stamp reuse. +The first file triggers a stamp purchase; subsequent files reuse that stamp. + +Usage: + python batch_upload.py --directory ./sample_files + python batch_upload.py --directory ./sample_files --std "PROV-STD-V1" +""" + +import argparse +import hashlib +import json +import os +import subprocess +import sys +from pathlib import Path + + +def sha256_file(path: str) -> str: + """Compute SHA-256 hash of a file.""" + h = hashlib.sha256() + with open(path, "rb") as f: + for chunk in iter(lambda: f.read(8192), b""): + h.update(chunk) + return h.hexdigest() + + +def run_cli(*args) -> subprocess.CompletedProcess: + """Run a swarm-prov-upload CLI command.""" + cmd = ["swarm-prov-upload"] + list(args) + result = subprocess.run(cmd, capture_output=True, text=True) + return result + + +def extract_swarm_ref(output: str) -> str: + """Extract Swarm reference hash from CLI output.""" + lines = output.splitlines() + for i, line in enumerate(lines): + if "Swarm Reference Hash:" in line and i + 1 < len(lines): + ref = lines[i + 1].strip() + if len(ref) >= 64: + return ref + return "" + + +def extract_stamp_id(output: str) -> str: + """Extract stamp ID from verbose CLI output. + + Looks for 'Stamp ID Received: ' in verbose output. + """ + for line in output.splitlines(): + if "Stamp ID Received:" in line: + parts = line.split("Stamp ID Received:") + if len(parts) > 1: + stamp_id = parts[1].strip() + if len(stamp_id) >= 16: + return stamp_id + return "" + + +def upload_file(file_path: str, std: str = None, stamp_id: str = None, + verbose: bool = False) -> dict: + """Upload a single file, optionally reusing a stamp. + + Returns dict with 'reference', 'stamp_id', and 'hash' keys. + """ + args = ["upload", "--file", file_path] + if std: + args.extend(["--std", std]) + if stamp_id: + args.extend(["--stamp-id", stamp_id]) + if verbose: + args.append("-v") + + if not stamp_id: + # Try pool first + result = run_cli(*(args + ["--usePool"])) + if result.returncode != 0: + result = run_cli(*args) + else: + result = run_cli(*args) + + if result.returncode != 0: + return {"error": result.stderr or result.stdout} + + output = result.stdout + "\n" + result.stderr + ref = extract_swarm_ref(result.stdout) + sid = extract_stamp_id(output) + + return { + "reference": ref, + "stamp_id": sid, + "hash": sha256_file(file_path), + } + + +def main(): + parser = argparse.ArgumentParser( + description="Batch upload files to Swarm with stamp reuse" + ) + parser.add_argument( + "--directory", "-d", + required=True, + help="Directory containing files to upload", + ) + parser.add_argument( + "--std", "-s", + default=None, + help="Provenance standard to apply (e.g., PROV-STD-V1)", + ) + parser.add_argument( + "--output", "-o", + default=None, + help="Output manifest file (default: /manifest.json)", + ) + args = parser.parse_args() + + if not os.path.isdir(args.directory): + print(f"ERROR: Directory not found: {args.directory}") + sys.exit(1) + + files = sorted([ + f for f in os.listdir(args.directory) + if os.path.isfile(os.path.join(args.directory, f)) + ]) + + if not files: + print(f"ERROR: No files found in {args.directory}") + sys.exit(1) + + print(f"Batch uploading {len(files)} files from {args.directory}") + if args.std: + print(f"Provenance standard: {args.std}") + + manifest = {} + stamp_id = None + + for i, filename in enumerate(files): + file_path = os.path.join(args.directory, filename) + print(f"\n[{i + 1}/{len(files)}] Uploading: {filename}") + + # First upload uses verbose to capture stamp ID + verbose = (i == 0 and stamp_id is None) + result = upload_file(file_path, std=args.std, stamp_id=stamp_id, + verbose=verbose) + + if "error" in result: + print(f" ERROR: {result['error']}") + sys.exit(1) + + if not result["reference"]: + print(" ERROR: Could not extract Swarm reference") + sys.exit(1) + + # Capture stamp ID from first upload for reuse + if stamp_id is None and result.get("stamp_id"): + stamp_id = result["stamp_id"] + print(f" Stamp ID captured: {stamp_id[:16]}...") + + manifest[filename] = { + "reference": result["reference"], + "content_hash": result["hash"], + } + print(f" Reference: {result['reference']}") + + # Save manifest + output_path = args.output or os.path.join(args.directory, "manifest.json") + with open(output_path, "w") as f: + json.dump(manifest, f, indent=2) + print(f"\nManifest saved: {output_path}") + print(f"Total files uploaded: {len(manifest)}") + + +if __name__ == "__main__": + main() diff --git a/examples/04-batch-processing/demo.sh b/examples/04-batch-processing/demo.sh new file mode 100755 index 0000000..32528aa --- /dev/null +++ b/examples/04-batch-processing/demo.sh @@ -0,0 +1,159 @@ +#!/usr/bin/env bash +# +# Batch Processing Demo +# +# Demonstrates uploading multiple files with stamp reuse: +# 1. Upload first file with --size medium -v to capture stamp ID +# 2. Upload remaining files with --stamp-id (reuses stamp, skips purchase) +# 3. Build a manifest of all uploaded files +# 4. Download and verify one file +# +# Usage: ./demo.sh + +set -euo pipefail + +SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)" +SAMPLE_DIR="$SCRIPT_DIR/sample_files" +DOWNLOAD_DIR="$SCRIPT_DIR/downloads" + +echo "======================================" +echo " Swarm Provenance CLI - Batch Upload" +echo "======================================" +echo + +# --- Step 0: Check CLI is installed --- +if ! command -v swarm-prov-upload &>/dev/null; then + echo "ERROR: swarm-prov-upload not found. Install with: pip install -e ." + exit 1 +fi + +echo "CLI version: $(swarm-prov-upload --version)" +echo + +# --- Step 1: Check gateway health --- +echo "--- Step 1: Check gateway health ---" +swarm-prov-upload health +echo + +# --- Step 2: Upload first file with verbose to capture stamp ID --- +echo "--- Step 2: Upload first file with --size medium -v ---" +FIRST_FILE="$SAMPLE_DIR/log_entry_001.json" +echo "Uploading: log_entry_001.json" +echo " SHA256: $(shasum -a 256 "$FIRST_FILE" | cut -d' ' -f1)" + +UPLOAD_OUTPUT=$(swarm-prov-upload upload --file "$FIRST_FILE" --size medium -v --usePool 2>&1) || { + echo " Pool not available, falling back to regular stamp purchase..." + UPLOAD_OUTPUT=$(swarm-prov-upload upload --file "$FIRST_FILE" --size medium -v 2>&1) +} + +FIRST_REF=$(echo "$UPLOAD_OUTPUT" | grep -A1 "Swarm Reference Hash:" | tail -1 | tr -d '[:space:]') +if [ -z "$FIRST_REF" ] || [ ${#FIRST_REF} -lt 64 ]; then + echo "ERROR: Could not extract Swarm reference" + echo "Raw output: $UPLOAD_OUTPUT" + exit 1 +fi + +# Extract stamp ID from verbose output +STAMP_ID=$(echo "$UPLOAD_OUTPUT" | grep "Stamp ID Received:" | awk -F'Stamp ID Received: ' '{print $2}' | tr -d '[:space:]') +if [ -z "$STAMP_ID" ] || [ ${#STAMP_ID} -lt 16 ]; then + echo "WARNING: Could not extract stamp ID from verbose output" + echo "Subsequent uploads will purchase new stamps" + USE_STAMP="" +else + echo " Stamp ID captured: ${STAMP_ID:0:16}..." + USE_STAMP="--stamp-id $STAMP_ID" +fi + +echo " Reference: $FIRST_REF" +echo + +# --- Step 3: Upload remaining files with stamp reuse --- +echo "--- Step 3: Upload remaining files with stamp reuse ---" +declare -A MANIFEST +MANIFEST["log_entry_001.json"]="$FIRST_REF" + +REMAINING=("log_entry_002.json" "log_entry_003.json") +for entry in "${REMAINING[@]}"; do + FILE="$SAMPLE_DIR/$entry" + echo "Uploading: $entry (reusing stamp)" + echo " SHA256: $(shasum -a 256 "$FILE" | cut -d' ' -f1)" + + if [ -n "$USE_STAMP" ]; then + # shellcheck disable=SC2086 + UPLOAD_OUTPUT=$(swarm-prov-upload upload --file "$FILE" $USE_STAMP 2>&1) + else + UPLOAD_OUTPUT=$(swarm-prov-upload upload --file "$FILE" --usePool 2>&1) || { + echo " Pool not available, falling back to regular stamp purchase..." + UPLOAD_OUTPUT=$(swarm-prov-upload upload --file "$FILE" 2>&1) + } + fi + + SWARM_REF=$(echo "$UPLOAD_OUTPUT" | grep -A1 "Swarm Reference Hash:" | tail -1 | tr -d '[:space:]') + + if [ -z "$SWARM_REF" ] || [ ${#SWARM_REF} -lt 64 ]; then + echo "ERROR: Could not extract Swarm reference for $entry" + echo "Raw output: $UPLOAD_OUTPUT" + exit 1 + fi + + MANIFEST["$entry"]="$SWARM_REF" + echo " Reference: $SWARM_REF" + echo +done + +echo "All ${#MANIFEST[@]} files uploaded." +echo + +# --- Step 4: Build manifest --- +echo "--- Step 4: Build manifest ---" +MANIFEST_FILE="$SCRIPT_DIR/manifest.json" +echo "{" > "$MANIFEST_FILE" +FIRST_ENTRY=true +for key in $(echo "${!MANIFEST[@]}" | tr ' ' '\n' | sort); do + if [ "$FIRST_ENTRY" = true ]; then + FIRST_ENTRY=false + else + echo "," >> "$MANIFEST_FILE" + fi + printf ' "%s": "%s"' "$key" "${MANIFEST[$key]}" >> "$MANIFEST_FILE" +done +echo "" >> "$MANIFEST_FILE" +echo "}" >> "$MANIFEST_FILE" +echo "Manifest saved: $MANIFEST_FILE" +cat "$MANIFEST_FILE" +echo + +# --- Step 5: Download and verify one file --- +echo "--- Step 5: Download and verify log_entry_001.json ---" +rm -rf "$DOWNLOAD_DIR" +mkdir -p "$DOWNLOAD_DIR" + +echo "Downloading: $FIRST_REF" +swarm-prov-upload download "$FIRST_REF" --output-dir "$DOWNLOAD_DIR" +echo + +# --- Step 6: Verify integrity --- +echo "--- Step 6: Compare SHA-256 hashes ---" +ORIGINAL_HASH=$(shasum -a 256 "$FIRST_FILE" | cut -d' ' -f1) +DOWNLOADED_FILE=$(ls "$DOWNLOAD_DIR"/*.data 2>/dev/null | head -1) +if [ -z "$DOWNLOADED_FILE" ]; then + DOWNLOADED_FILE=$(ls "$DOWNLOAD_DIR"/ | head -1) + DOWNLOADED_FILE="$DOWNLOAD_DIR/$DOWNLOADED_FILE" +fi +DOWNLOADED_HASH=$(shasum -a 256 "$DOWNLOADED_FILE" | cut -d' ' -f1) + +echo "Original: $ORIGINAL_HASH" +echo "Downloaded: $DOWNLOADED_HASH" +echo + +if [ "$ORIGINAL_HASH" = "$DOWNLOADED_HASH" ]; then + echo "PASS: Batch upload integrity verified - hashes match." +else + echo "FAIL: Hash mismatch - data integrity compromised!" + exit 1 +fi + +echo +echo "--- Demo complete ---" +echo "Uploaded ${#MANIFEST[@]} files with stamp reuse." +echo "Manifest: $MANIFEST_FILE" diff --git a/examples/04-batch-processing/run_demo.py b/examples/04-batch-processing/run_demo.py new file mode 100644 index 0000000..4c16d98 --- /dev/null +++ b/examples/04-batch-processing/run_demo.py @@ -0,0 +1,216 @@ +#!/usr/bin/env python3 +""" +Batch Processing Demo - Python Version + +Demonstrates uploading multiple files with stamp reuse: +1. Upload first file with --size medium -v to capture stamp ID +2. Upload remaining files with --stamp-id (reuses stamp, skips purchase) +3. Build a manifest of all uploaded files +4. Download and verify one file + +Usage: + python run_demo.py + python run_demo.py --files log_entry_001.json log_entry_002.json log_entry_003.json +""" + +import argparse +import hashlib +import json +import os +import subprocess +import sys +from pathlib import Path + +SCRIPT_DIR = Path(__file__).parent + +DEFAULT_FILES = [ + "sample_files/log_entry_001.json", + "sample_files/log_entry_002.json", + "sample_files/log_entry_003.json", +] + + +def sha256_file(path: str) -> str: + """Compute SHA-256 hash of a file.""" + h = hashlib.sha256() + with open(path, "rb") as f: + for chunk in iter(lambda: f.read(8192), b""): + h.update(chunk) + return h.hexdigest() + + +def run_cli(*args) -> subprocess.CompletedProcess: + """Run a swarm-prov-upload CLI command.""" + cmd = ["swarm-prov-upload"] + list(args) + result = subprocess.run(cmd, capture_output=True, text=True) + return result + + +def extract_swarm_ref(output: str) -> str: + """Extract Swarm reference hash from CLI output.""" + lines = output.splitlines() + for i, line in enumerate(lines): + if "Swarm Reference Hash:" in line and i + 1 < len(lines): + ref = lines[i + 1].strip() + if len(ref) >= 64: + return ref + return "" + + +def extract_stamp_id(output: str) -> str: + """Extract stamp ID from verbose CLI output.""" + for line in output.splitlines(): + if "Stamp ID Received:" in line: + parts = line.split("Stamp ID Received:") + if len(parts) > 1: + stamp_id = parts[1].strip() + if len(stamp_id) >= 16: + return stamp_id + return "" + + +def main(): + parser = argparse.ArgumentParser(description="Batch processing demo") + parser.add_argument( + "--files", "-f", + nargs="+", + default=DEFAULT_FILES, + help="Files to upload (default: 3 sample log entries)", + ) + args = parser.parse_args() + + print("=" * 50) + print(" Swarm Provenance CLI - Batch Upload (Python)") + print("=" * 50) + + # --- Step 1: Check health --- + print("\n--- Step 1: Check gateway health ---") + result = run_cli("health") + if result.returncode != 0: + print(f"Gateway not available: {result.stderr or result.stdout}") + sys.exit(1) + print(result.stdout.strip()) + + # Verify all files exist + for file_rel in args.files: + file_path = str(SCRIPT_DIR / file_rel) + if not os.path.exists(file_path): + print(f"ERROR: File not found: {file_path}") + sys.exit(1) + + # --- Step 2: Upload first file with verbose to capture stamp --- + print("\n--- Step 2: Upload first file with --size medium -v ---") + first_file = str(SCRIPT_DIR / args.files[0]) + first_name = os.path.basename(args.files[0]) + original_hash = sha256_file(first_file) + print(f"Uploading: {first_name}") + print(f" SHA256: {original_hash}") + + result = run_cli("upload", "--file", first_file, "--size", "medium", "-v", "--usePool") + if result.returncode != 0: + print(" Pool not available, falling back to regular stamp purchase...") + result = run_cli("upload", "--file", first_file, "--size", "medium", "-v") + if result.returncode != 0: + print(f" Upload failed: {result.stderr or result.stdout}") + sys.exit(1) + + first_ref = extract_swarm_ref(result.stdout) + if not first_ref: + print(" Could not extract Swarm reference from output") + sys.exit(1) + + combined_output = result.stdout + "\n" + result.stderr + stamp_id = extract_stamp_id(combined_output) + if stamp_id: + print(f" Stamp ID captured: {stamp_id[:16]}...") + else: + print(" WARNING: Could not extract stamp ID from verbose output") + + print(f" Reference: {first_ref}") + + manifest = {first_name: first_ref} + + # --- Step 3: Upload remaining files with stamp reuse --- + print("\n--- Step 3: Upload remaining files with stamp reuse ---") + for file_rel in args.files[1:]: + file_path = str(SCRIPT_DIR / file_rel) + filename = os.path.basename(file_rel) + print(f"\nUploading: {filename}") + print(f" SHA256: {sha256_file(file_path)}") + + if stamp_id: + print(" (reusing stamp)") + result = run_cli("upload", "--file", file_path, "--stamp-id", stamp_id) + else: + result = run_cli("upload", "--file", file_path, "--usePool") + if result.returncode != 0: + print(" Pool not available, falling back to regular stamp purchase...") + result = run_cli("upload", "--file", file_path) + + if result.returncode != 0: + print(f" Upload failed: {result.stderr or result.stdout}") + sys.exit(1) + + swarm_ref = extract_swarm_ref(result.stdout) + if not swarm_ref: + print(" Could not extract Swarm reference") + sys.exit(1) + + manifest[filename] = swarm_ref + print(f" Reference: {swarm_ref}") + + print(f"\nAll {len(manifest)} files uploaded.") + + # --- Step 4: Build manifest --- + print("\n--- Step 4: Build manifest ---") + manifest_path = str(SCRIPT_DIR / "manifest.json") + with open(manifest_path, "w") as f: + json.dump(manifest, f, indent=2) + print(f"Manifest saved: {manifest_path}") + print(json.dumps(manifest, indent=2)) + + # --- Step 5: Download and verify first file --- + print(f"\n--- Step 5: Download and verify {first_name} ---") + download_dir = str(SCRIPT_DIR / "downloads") + os.makedirs(download_dir, exist_ok=True) + for f in os.listdir(download_dir): + os.remove(os.path.join(download_dir, f)) + + result = run_cli("download", first_ref, "--output-dir", download_dir) + if result.returncode != 0: + print(f"Download failed: {result.stderr or result.stdout}") + sys.exit(1) + print(result.stdout.strip()) + + # --- Step 6: Verify integrity --- + print("\n--- Step 6: Verify integrity ---") + downloaded_files = os.listdir(download_dir) + if not downloaded_files: + print("ERROR: No files in download directory") + sys.exit(1) + + data_files = [f for f in downloaded_files if f.endswith(".data")] + if data_files: + downloaded_file = os.path.join(download_dir, data_files[0]) + else: + downloaded_file = os.path.join(download_dir, downloaded_files[0]) + + downloaded_hash = sha256_file(downloaded_file) + + print(f"Original: {original_hash}") + print(f"Downloaded: {downloaded_hash}") + + if original_hash == downloaded_hash: + print("\nPASS: Batch upload integrity verified - hashes match.") + else: + print("\nFAIL: Hash mismatch - data integrity compromised!") + sys.exit(1) + + # --- Summary --- + print("\n--- Summary ---") + print(f"Uploaded {len(manifest)} files with stamp reuse.") + print(f"Manifest: {manifest_path}") + + +if __name__ == "__main__": + main() diff --git a/examples/04-batch-processing/sample_files/log_entry_001.json b/examples/04-batch-processing/sample_files/log_entry_001.json new file mode 100644 index 0000000..d080c54 --- /dev/null +++ b/examples/04-batch-processing/sample_files/log_entry_001.json @@ -0,0 +1,18 @@ +{ + "log_id": "LOG-2024-001", + "timestamp": "2024-01-15T09:30:00Z", + "level": "INFO", + "service": "auth-gateway", + "event": "user_login", + "details": { + "user_id": "usr_7a3b2c", + "ip_address": "192.168.1.42", + "method": "oauth2", + "success": true + }, + "metadata": { + "region": "eu-west-1", + "version": "2.4.1", + "correlation_id": "corr-abc123" + } +} diff --git a/examples/04-batch-processing/sample_files/log_entry_002.json b/examples/04-batch-processing/sample_files/log_entry_002.json new file mode 100644 index 0000000..c0e462d --- /dev/null +++ b/examples/04-batch-processing/sample_files/log_entry_002.json @@ -0,0 +1,18 @@ +{ + "log_id": "LOG-2024-002", + "timestamp": "2024-01-15T09:31:15Z", + "level": "WARNING", + "service": "data-pipeline", + "event": "rate_limit_approached", + "details": { + "endpoint": "/api/v2/ingest", + "current_rate": 850, + "limit": 1000, + "window_seconds": 60 + }, + "metadata": { + "region": "eu-west-1", + "version": "3.1.0", + "correlation_id": "corr-def456" + } +} diff --git a/examples/04-batch-processing/sample_files/log_entry_003.json b/examples/04-batch-processing/sample_files/log_entry_003.json new file mode 100644 index 0000000..a01e6fb --- /dev/null +++ b/examples/04-batch-processing/sample_files/log_entry_003.json @@ -0,0 +1,19 @@ +{ + "log_id": "LOG-2024-003", + "timestamp": "2024-01-15T09:32:45Z", + "level": "ERROR", + "service": "payment-processor", + "event": "transaction_failed", + "details": { + "transaction_id": "txn_9x8y7z", + "amount": 149.99, + "currency": "EUR", + "error_code": "INSUFFICIENT_FUNDS", + "retry_count": 2 + }, + "metadata": { + "region": "eu-west-1", + "version": "1.8.3", + "correlation_id": "corr-ghi789" + } +} diff --git a/examples/07-stamp-management/README.md b/examples/07-stamp-management/README.md new file mode 100644 index 0000000..a29c717 --- /dev/null +++ b/examples/07-stamp-management/README.md @@ -0,0 +1,105 @@ +# Example 07: Stamp Management + +Demonstrates the full postage stamp lifecycle using the stamps subcommands. + +## What This Demonstrates + +- Checking stamp pool availability (`stamps pool-status`) +- Uploading a file with verbose output to capture a stamp ID +- Listing all stamps (`stamps list`) +- Inspecting stamp details (`stamps info `) +- Health-checking a stamp (`stamps check `) +- Extending a stamp (`stamps extend ` — requires funded wallet) + +## Use Case + +Postage stamps are the payment mechanism for Swarm storage. Understanding how to manage stamps is essential for production deployments: + +- **Pool stamps** are pre-purchased and shared, offering fast (~5s) uploads +- **Individual stamps** give you full control over capacity and duration +- **Monitoring** stamp utilization prevents unexpected expiration +- **Extending** stamps keeps your data available longer + +## Prerequisites + +1. Install the CLI: `pip install -e .` +2. Gateway access (default, no setup needed) +3. For `stamps extend`: a funded wallet with BZZ tokens (optional) + +## Quick Start + +### Shell + +```bash +chmod +x demo.sh +./demo.sh +``` + +### Python + +```bash +python run_demo.py +``` + +## Step-by-Step Walkthrough + +### 1. Check pool availability + +```bash +swarm-prov-upload stamps pool-status +``` + +Shows whether the stamp pool is enabled and how many stamps are available. + +### 2. Upload to acquire a stamp + +```bash +swarm-prov-upload upload --file sample_data.txt -v --usePool +``` + +The `-v` flag shows the stamp ID in the output. + +### 3. List all stamps + +```bash +swarm-prov-upload stamps list +``` + +### 4. Inspect stamp details + +```bash +swarm-prov-upload stamps info +``` + +Shows depth, amount, utilization, and TTL. + +### 5. Health-check a stamp + +```bash +swarm-prov-upload stamps check +``` + +### 6. Extend a stamp (optional) + +```bash +swarm-prov-upload stamps extend --amount 1000000 +``` + +Requires a funded wallet. This tops up the stamp to extend its lifetime. + +## Stamp Lifecycle Helper + +The `stamp_lifecycle.py` script runs the full lifecycle: + +```bash +python stamp_lifecycle.py --file sample_data.txt +``` + +## Files + +| File | Description | +|------|-------------| +| `demo.sh` | Shell demo — full stamp lifecycle | +| `run_demo.py` | Python demo — same workflow with argparse support | +| `stamp_lifecycle.py` | Standalone stamp lifecycle management tool | +| `sample_data.txt` | Sample file for triggering stamp creation | diff --git a/examples/07-stamp-management/demo.sh b/examples/07-stamp-management/demo.sh new file mode 100755 index 0000000..6677720 --- /dev/null +++ b/examples/07-stamp-management/demo.sh @@ -0,0 +1,96 @@ +#!/usr/bin/env bash +# +# Stamp Management Demo +# +# Demonstrates the full postage stamp lifecycle: +# 1. Check stamp pool availability +# 2. Upload a file with -v to capture stamp ID +# 3. List all stamps +# 4. Inspect stamp details +# 5. Health-check a stamp +# +# Usage: ./demo.sh + +set -euo pipefail + +SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)" +SAMPLE_FILE="$SCRIPT_DIR/sample_data.txt" + +echo "=============================================" +echo " Swarm Provenance CLI - Stamp Management" +echo "=============================================" +echo + +# --- Step 0: Check CLI is installed --- +if ! command -v swarm-prov-upload &>/dev/null; then + echo "ERROR: swarm-prov-upload not found. Install with: pip install -e ." + exit 1 +fi + +echo "CLI version: $(swarm-prov-upload --version)" +echo + +# --- Step 1: Check gateway health --- +echo "--- Step 1: Check gateway health ---" +swarm-prov-upload health +echo + +# --- Step 2: Check stamp pool status --- +echo "--- Step 2: Check stamp pool availability ---" +swarm-prov-upload stamps pool-status || echo "(pool status check returned non-zero)" +echo + +# --- Step 3: Upload file with verbose to capture stamp ID --- +echo "--- Step 3: Upload sample file with -v to capture stamp ID ---" +echo "Uploading: sample_data.txt" + +UPLOAD_OUTPUT=$(swarm-prov-upload upload --file "$SAMPLE_FILE" -v --usePool 2>&1) || { + echo " Pool not available, falling back to regular stamp purchase..." + UPLOAD_OUTPUT=$(swarm-prov-upload upload --file "$SAMPLE_FILE" -v 2>&1) +} + +SWARM_REF=$(echo "$UPLOAD_OUTPUT" | grep -A1 "Swarm Reference Hash:" | tail -1 | tr -d '[:space:]') +if [ -z "$SWARM_REF" ] || [ ${#SWARM_REF} -lt 64 ]; then + echo "ERROR: Could not extract Swarm reference" + echo "Raw output: $UPLOAD_OUTPUT" + exit 1 +fi + +STAMP_ID=$(echo "$UPLOAD_OUTPUT" | grep "Stamp ID Received:" | awk -F'Stamp ID Received: ' '{print $2}' | tr -d '[:space:]') +echo " Swarm reference: $SWARM_REF" + +if [ -z "$STAMP_ID" ] || [ ${#STAMP_ID} -lt 16 ]; then + echo "WARNING: Could not extract stamp ID from verbose output" + echo "Stamp lifecycle commands require a stamp ID." + echo + echo "PASS: Upload succeeded. Stamp lifecycle steps skipped (no stamp ID)." + exit 0 +fi + +echo " Stamp ID: $STAMP_ID" +echo + +# --- Step 4: List all stamps --- +echo "--- Step 4: List all stamps ---" +swarm-prov-upload stamps list || echo "(stamps list returned non-zero)" +echo + +# --- Step 5: Stamp info --- +echo "--- Step 5: Stamp details ---" +swarm-prov-upload stamps info "$STAMP_ID" || echo "(stamps info returned non-zero)" +echo + +# --- Step 6: Stamp health check --- +echo "--- Step 6: Stamp health check ---" +swarm-prov-upload stamps check "$STAMP_ID" || echo "(stamps check returned non-zero)" +echo + +echo "PASS: Stamp lifecycle demo complete." +echo +echo "--- Summary ---" +echo "Swarm reference: $SWARM_REF" +echo "Stamp ID: $STAMP_ID" +echo +echo "Additional stamp commands available:" +echo " swarm-prov-upload stamps extend $STAMP_ID --amount 1000000" +echo " (requires funded wallet with BZZ tokens)" diff --git a/examples/07-stamp-management/run_demo.py b/examples/07-stamp-management/run_demo.py new file mode 100644 index 0000000..3742897 --- /dev/null +++ b/examples/07-stamp-management/run_demo.py @@ -0,0 +1,148 @@ +#!/usr/bin/env python3 +""" +Stamp Management Demo - Python Version + +Demonstrates the full postage stamp lifecycle: +1. Check stamp pool availability +2. Upload a file with -v to capture stamp ID +3. List all stamps +4. Inspect stamp details +5. Health-check a stamp + +Usage: + python run_demo.py + python run_demo.py --file sample_data.txt +""" + +import argparse +import hashlib +import os +import subprocess +import sys +from pathlib import Path + +SCRIPT_DIR = Path(__file__).parent + + +def sha256_file(path: str) -> str: + """Compute SHA-256 hash of a file.""" + h = hashlib.sha256() + with open(path, "rb") as f: + for chunk in iter(lambda: f.read(8192), b""): + h.update(chunk) + return h.hexdigest() + + +def run_cli(*args) -> subprocess.CompletedProcess: + """Run a swarm-prov-upload CLI command.""" + cmd = ["swarm-prov-upload"] + list(args) + result = subprocess.run(cmd, capture_output=True, text=True) + return result + + +def extract_swarm_ref(output: str) -> str: + """Extract Swarm reference hash from CLI output.""" + lines = output.splitlines() + for i, line in enumerate(lines): + if "Swarm Reference Hash:" in line and i + 1 < len(lines): + ref = lines[i + 1].strip() + if len(ref) >= 64: + return ref + return "" + + +def extract_stamp_id(output: str) -> str: + """Extract stamp ID from verbose CLI output.""" + for line in output.splitlines(): + if "Stamp ID Received:" in line: + parts = line.split("Stamp ID Received:") + if len(parts) > 1: + stamp_id = parts[1].strip() + if len(stamp_id) >= 16: + return stamp_id + return "" + + +def main(): + parser = argparse.ArgumentParser(description="Stamp management demo") + parser.add_argument( + "--file", "-f", + default="sample_data.txt", + help="File to upload for stamp acquisition (default: sample_data.txt)", + ) + args = parser.parse_args() + + print("=" * 55) + print(" Swarm Provenance CLI - Stamp Management (Python)") + print("=" * 55) + + # --- Step 1: Check health --- + print("\n--- Step 1: Check gateway health ---") + result = run_cli("health") + if result.returncode != 0: + print(f"Gateway not available: {result.stderr or result.stdout}") + sys.exit(1) + print(result.stdout.strip()) + + # --- Step 2: Check stamp pool --- + print("\n--- Step 2: Check stamp pool availability ---") + result = run_cli("stamps", "pool-status") + print(result.stdout.strip() if result.stdout else result.stderr.strip()) + + # --- Step 3: Upload file with verbose --- + print("\n--- Step 3: Upload file with -v to capture stamp ID ---") + file_path = str(SCRIPT_DIR / args.file) + if not os.path.exists(file_path): + print(f"ERROR: File not found: {file_path}") + sys.exit(1) + + print(f"Uploading: {args.file}") + result = run_cli("upload", "--file", file_path, "-v", "--usePool") + if result.returncode != 0: + print(" Pool not available, falling back to regular stamp purchase...") + result = run_cli("upload", "--file", file_path, "-v") + if result.returncode != 0: + print(f" Upload failed: {result.stderr or result.stdout}") + sys.exit(1) + + swarm_ref = extract_swarm_ref(result.stdout) + if not swarm_ref: + print(" Could not extract Swarm reference") + sys.exit(1) + + combined_output = result.stdout + "\n" + result.stderr + stamp_id = extract_stamp_id(combined_output) + + print(f" Swarm reference: {swarm_ref}") + + if not stamp_id: + print(" WARNING: Could not extract stamp ID from verbose output") + print(" Stamp lifecycle commands require a stamp ID.") + print("\nPASS: Upload succeeded. Stamp lifecycle steps skipped.") + return + + print(f" Stamp ID: {stamp_id}") + + # --- Step 4: List stamps --- + print("\n--- Step 4: List all stamps ---") + result = run_cli("stamps", "list") + print(result.stdout.strip() if result.stdout else result.stderr.strip()) + + # --- Step 5: Stamp info --- + print(f"\n--- Step 5: Stamp details for {stamp_id[:16]}... ---") + result = run_cli("stamps", "info", stamp_id) + print(result.stdout.strip() if result.stdout else result.stderr.strip()) + + # --- Step 6: Stamp health check --- + print("\n--- Step 6: Stamp health check ---") + result = run_cli("stamps", "check", stamp_id) + print(result.stdout.strip() if result.stdout else result.stderr.strip()) + + print("\nPASS: Stamp lifecycle demo complete.") + print(f"\n--- Summary ---") + print(f"Swarm reference: {swarm_ref}") + print(f"Stamp ID: {stamp_id}") + + +if __name__ == "__main__": + main() diff --git a/examples/07-stamp-management/sample_data.txt b/examples/07-stamp-management/sample_data.txt new file mode 100644 index 0000000..d6b2bd7 --- /dev/null +++ b/examples/07-stamp-management/sample_data.txt @@ -0,0 +1,20 @@ +Stamp Management Demo - Sample Data +===================================== + +This file is used to demonstrate postage stamp lifecycle management. + +When uploaded to Swarm, a postage stamp is purchased (or acquired from the pool) +to pay for storage. This stamp has properties like: + +- Batch ID: Unique identifier for the stamp +- Depth: Determines the number of chunks covered +- Amount: BZZ tokens locked for storage +- Utilization: How much of the stamp capacity is used +- TTL: Time-to-live before the stamp expires + +Use the stamps subcommands to inspect and manage your stamps: + stamps list - Show all stamps + stamps info - Detailed stamp information + stamps check - Health check a stamp + stamps extend --amount - Top up a stamp + stamps pool-status - Check the stamp pool diff --git a/examples/07-stamp-management/stamp_lifecycle.py b/examples/07-stamp-management/stamp_lifecycle.py new file mode 100644 index 0000000..6614f43 --- /dev/null +++ b/examples/07-stamp-management/stamp_lifecycle.py @@ -0,0 +1,167 @@ +#!/usr/bin/env python3 +""" +Stamp Lifecycle Manager + +Demonstrates the full postage stamp lifecycle: +1. Check pool availability +2. Upload a file to acquire a stamp +3. List all stamps +4. Inspect stamp details +5. Health-check a stamp +6. Attempt to extend a stamp (may require funded wallet) + +Usage: + python stamp_lifecycle.py + python stamp_lifecycle.py --file sample_data.txt +""" + +import argparse +import json +import os +import subprocess +import sys +from pathlib import Path + + +def run_cli(*args) -> subprocess.CompletedProcess: + """Run a swarm-prov-upload CLI command.""" + cmd = ["swarm-prov-upload"] + list(args) + result = subprocess.run(cmd, capture_output=True, text=True) + return result + + +def extract_stamp_id(output: str) -> str: + """Extract stamp ID from verbose CLI output.""" + for line in output.splitlines(): + if "Stamp ID Received:" in line: + parts = line.split("Stamp ID Received:") + if len(parts) > 1: + stamp_id = parts[1].strip() + if len(stamp_id) >= 16: + return stamp_id + return "" + + +def extract_swarm_ref(output: str) -> str: + """Extract Swarm reference hash from CLI output.""" + lines = output.splitlines() + for i, line in enumerate(lines): + if "Swarm Reference Hash:" in line and i + 1 < len(lines): + ref = lines[i + 1].strip() + if len(ref) >= 64: + return ref + return "" + + +def pool_status(): + """Check stamp pool availability.""" + result = run_cli("stamps", "pool-status") + print(result.stdout.strip() if result.stdout else result.stderr.strip()) + return result.returncode == 0 + + +def list_stamps(): + """List all stamps.""" + result = run_cli("stamps", "list") + print(result.stdout.strip() if result.stdout else result.stderr.strip()) + return result.returncode == 0 + + +def stamp_info(stamp_id: str): + """Get detailed stamp information.""" + result = run_cli("stamps", "info", stamp_id) + print(result.stdout.strip() if result.stdout else result.stderr.strip()) + return result.returncode == 0 + + +def stamp_check(stamp_id: str): + """Health-check a stamp.""" + result = run_cli("stamps", "check", stamp_id) + print(result.stdout.strip() if result.stdout else result.stderr.strip()) + return result.returncode == 0 + + +def stamp_extend(stamp_id: str, amount: int = 1000000): + """Attempt to extend a stamp (requires funded wallet).""" + result = run_cli("stamps", "extend", stamp_id, "--amount", str(amount)) + print(result.stdout.strip() if result.stdout else result.stderr.strip()) + return result.returncode == 0 + + +def main(): + parser = argparse.ArgumentParser(description="Stamp lifecycle demo") + parser.add_argument( + "--file", "-f", + default=None, + help="File to upload for stamp acquisition", + ) + args = parser.parse_args() + + print("=" * 55) + print(" Stamp Lifecycle Manager") + print("=" * 55) + + # Step 1: Pool status + print("\n--- Step 1: Check stamp pool availability ---") + pool_status() + + # Step 2: Upload to acquire stamp + if args.file: + print(f"\n--- Step 2: Upload file to acquire stamp ---") + if not os.path.exists(args.file): + print(f"ERROR: File not found: {args.file}") + sys.exit(1) + + result = run_cli("upload", "--file", args.file, "-v", "--usePool") + if result.returncode != 0: + print("Pool not available, falling back to regular stamp purchase...") + result = run_cli("upload", "--file", args.file, "-v") + if result.returncode != 0: + print(f"Upload failed: {result.stderr or result.stdout}") + sys.exit(1) + + output = result.stdout + "\n" + result.stderr + stamp_id = extract_stamp_id(output) + swarm_ref = extract_swarm_ref(result.stdout) + + if not stamp_id: + print("WARNING: Could not extract stamp ID from verbose output") + print("Continuing with stamps list to find stamps...") + else: + print(f"Stamp acquired: {stamp_id}") + + if swarm_ref: + print(f"Swarm reference: {swarm_ref}") + else: + stamp_id = None + print("\n--- Step 2: Skipped (no --file provided) ---") + + # Step 3: List stamps + print("\n--- Step 3: List all stamps ---") + list_stamps() + + # Step 4: Stamp info (if we have a stamp ID) + if stamp_id: + print(f"\n--- Step 4: Stamp details for {stamp_id[:16]}... ---") + stamp_info(stamp_id) + + # Step 5: Stamp health check + print(f"\n--- Step 5: Stamp health check ---") + stamp_check(stamp_id) + + # Step 6: Attempt extend (may fail without funded wallet) + print(f"\n--- Step 6: Attempt stamp extension ---") + print("Note: Extension requires a funded wallet with BZZ tokens.") + ok = stamp_extend(stamp_id) + if not ok: + print("Extension failed (expected if wallet is not funded).") + print("This is normal for demo environments.") + else: + print("\n--- Steps 4-6: Skipped (no stamp ID available) ---") + print("Provide --file to upload and acquire a stamp for full lifecycle demo.") + + print("\n--- Lifecycle demo complete ---") + + +if __name__ == "__main__": + main() diff --git a/examples/08-ci-cd-integration/README.md b/examples/08-ci-cd-integration/README.md new file mode 100644 index 0000000..02cff64 --- /dev/null +++ b/examples/08-ci-cd-integration/README.md @@ -0,0 +1,101 @@ +# Example 08: CI/CD Integration + +Demonstrates archiving build artifacts to Swarm from CI/CD pipelines. + +## What This Demonstrates + +- Uploading build artifacts with `--std "CI-ARTIFACT-V1"` +- Saving archive receipts (JSON manifests with references, hashes, timestamps) +- Downloading and verifying archived artifacts +- Sample GitHub Actions and GitLab CI configurations + +## Use Case + +CI/CD pipelines produce build artifacts that need immutable archival for: + +- **Compliance**: Prove exactly what was built and deployed +- **Reproducibility**: Retrieve any historical build artifact by its Swarm reference +- **Audit trails**: Track which artifacts were deployed to production +- **Supply chain security**: Verify artifact integrity before deployment + +By archiving artifacts to Swarm with a provenance standard, each build gets a permanent, content-addressed record. + +## Prerequisites + +1. Install the CLI: `pip install -e .` +2. Gateway access (default, no setup needed) + +## Quick Start + +### Shell + +```bash +chmod +x demo.sh +./demo.sh +``` + +### Python + +```bash +python run_demo.py +``` + +## Step-by-Step Walkthrough + +### 1. Upload build artifacts + +```bash +swarm-prov-upload upload --file sample_artifacts/build_info.json --std "CI-ARTIFACT-V1" --usePool +swarm-prov-upload upload --file sample_artifacts/release_notes.txt --std "CI-ARTIFACT-V1" --usePool +``` + +### 2. Save archive receipt + +The demo saves a JSON receipt with references, content hashes, and timestamps for each artifact. + +### 3. Download and verify + +```bash +swarm-prov-upload download --output-dir ./downloads +``` + +Compare SHA-256 hashes to verify the artifact is intact. + +## CI/CD Configuration Templates + +### GitHub Actions + +See `github-action.yml` for a sample workflow that archives artifacts after a successful build. Copy to `.github/workflows/` in your repository. + +### GitLab CI + +See `gitlab-ci.yml` for a sample pipeline configuration. Copy the relevant sections to your `.gitlab-ci.yml`. + +Both templates use `swarm-prov-upload` with `--std "CI-ARTIFACT-V1"` and `--usePool` for fast uploads. + +## Archive Artifacts Helper + +The `archive_artifacts.py` script automates artifact archival: + +```bash +python archive_artifacts.py --directory ./dist +python archive_artifacts.py --directory ./dist --std "CI-ARTIFACT-V1" +``` + +## Sample Artifacts + +| File | Description | +|------|-------------| +| `sample_artifacts/build_info.json` | Build metadata (version, commit, test results) | +| `sample_artifacts/release_notes.txt` | Release notes for v2.1.0 | + +## Files + +| File | Description | +|------|-------------| +| `demo.sh` | Shell demo — archive and verify artifacts | +| `run_demo.py` | Python demo — same workflow with argparse support | +| `archive_artifacts.py` | Standalone artifact archival tool | +| `github-action.yml` | Sample GitHub Actions workflow | +| `gitlab-ci.yml` | Sample GitLab CI configuration | +| `sample_artifacts/` | Directory of sample build artifacts | diff --git a/examples/08-ci-cd-integration/archive_artifacts.py b/examples/08-ci-cd-integration/archive_artifacts.py new file mode 100644 index 0000000..24f57ad --- /dev/null +++ b/examples/08-ci-cd-integration/archive_artifacts.py @@ -0,0 +1,148 @@ +#!/usr/bin/env python3 +""" +CI/CD Artifact Archiver + +Archives build artifacts to Swarm with provenance metadata. +Designed for use in CI/CD pipelines (GitHub Actions, GitLab CI, etc.). + +Usage: + python archive_artifacts.py --directory ./sample_artifacts + python archive_artifacts.py --directory ./dist --std "CI-ARTIFACT-V1" +""" + +import argparse +import hashlib +import json +import os +import subprocess +import sys +from datetime import datetime, timezone +from pathlib import Path + + +def sha256_file(path: str) -> str: + """Compute SHA-256 hash of a file.""" + h = hashlib.sha256() + with open(path, "rb") as f: + for chunk in iter(lambda: f.read(8192), b""): + h.update(chunk) + return h.hexdigest() + + +def run_cli(*args) -> subprocess.CompletedProcess: + """Run a swarm-prov-upload CLI command.""" + cmd = ["swarm-prov-upload"] + list(args) + result = subprocess.run(cmd, capture_output=True, text=True) + return result + + +def extract_swarm_ref(output: str) -> str: + """Extract Swarm reference hash from CLI output.""" + lines = output.splitlines() + for i, line in enumerate(lines): + if "Swarm Reference Hash:" in line and i + 1 < len(lines): + ref = lines[i + 1].strip() + if len(ref) >= 64: + return ref + return "" + + +def archive_file(file_path: str, std: str = None) -> dict: + """Upload a single artifact and return its receipt. + + Returns dict with reference, hash, filename, and timestamp. + """ + args = ["upload", "--file", file_path] + if std: + args.extend(["--std", std]) + + result = run_cli(*(args + ["--usePool"])) + if result.returncode != 0: + print(f" Pool not available, falling back to regular stamp purchase...") + result = run_cli(*args) + + if result.returncode != 0: + return {"error": result.stderr or result.stdout} + + ref = extract_swarm_ref(result.stdout) + return { + "filename": os.path.basename(file_path), + "reference": ref, + "content_hash": sha256_file(file_path), + "size_bytes": os.path.getsize(file_path), + "archived_at": datetime.now(timezone.utc).isoformat(), + } + + +def main(): + parser = argparse.ArgumentParser( + description="Archive build artifacts to Swarm" + ) + parser.add_argument( + "--directory", "-d", + required=True, + help="Directory containing artifacts to archive", + ) + parser.add_argument( + "--std", "-s", + default="CI-ARTIFACT-V1", + help="Provenance standard (default: CI-ARTIFACT-V1)", + ) + parser.add_argument( + "--output", "-o", + default=None, + help="Output receipt file (default: /archive_receipt.json)", + ) + args = parser.parse_args() + + if not os.path.isdir(args.directory): + print(f"ERROR: Directory not found: {args.directory}") + sys.exit(1) + + files = sorted([ + f for f in os.listdir(args.directory) + if os.path.isfile(os.path.join(args.directory, f)) + ]) + + if not files: + print(f"ERROR: No files found in {args.directory}") + sys.exit(1) + + print(f"Archiving {len(files)} artifacts from {args.directory}") + print(f"Provenance standard: {args.std}") + + receipt = { + "pipeline": "ci-cd-archive", + "timestamp": datetime.now(timezone.utc).isoformat(), + "standard": args.std, + "artifacts": [], + } + + for i, filename in enumerate(files): + file_path = os.path.join(args.directory, filename) + print(f"\n[{i + 1}/{len(files)}] Archiving: {filename}") + + result = archive_file(file_path, std=args.std) + + if "error" in result: + print(f" ERROR: {result['error']}") + sys.exit(1) + + if not result["reference"]: + print(" ERROR: Could not extract Swarm reference") + sys.exit(1) + + receipt["artifacts"].append(result) + print(f" Reference: {result['reference']}") + print(f" Hash: {result['content_hash']}") + + # Save receipt + output_path = args.output or os.path.join(args.directory, "archive_receipt.json") + with open(output_path, "w") as f: + json.dump(receipt, f, indent=2) + print(f"\nArchive receipt saved: {output_path}") + print(f"Total artifacts archived: {len(receipt['artifacts'])}") + + +if __name__ == "__main__": + main() diff --git a/examples/08-ci-cd-integration/demo.sh b/examples/08-ci-cd-integration/demo.sh new file mode 100755 index 0000000..0571a5c --- /dev/null +++ b/examples/08-ci-cd-integration/demo.sh @@ -0,0 +1,135 @@ +#!/usr/bin/env bash +# +# CI/CD Integration Demo +# +# Demonstrates archiving build artifacts to Swarm: +# 1. Upload build artifacts with --std "CI-ARTIFACT-V1" +# 2. Save receipt manifest with references and hashes +# 3. Download and verify one artifact +# +# Usage: ./demo.sh + +set -euo pipefail + +SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)" +ARTIFACTS_DIR="$SCRIPT_DIR/sample_artifacts" +DOWNLOAD_DIR="$SCRIPT_DIR/downloads" + +echo "=============================================" +echo " Swarm Provenance CLI - CI/CD Integration" +echo "=============================================" +echo + +# --- Step 0: Check CLI is installed --- +if ! command -v swarm-prov-upload &>/dev/null; then + echo "ERROR: swarm-prov-upload not found. Install with: pip install -e ." + exit 1 +fi + +echo "CLI version: $(swarm-prov-upload --version)" +echo + +# --- Step 1: Check gateway health --- +echo "--- Step 1: Check gateway health ---" +swarm-prov-upload health +echo + +# --- Step 2: Upload build artifacts --- +echo "--- Step 2: Upload build artifacts with --std CI-ARTIFACT-V1 ---" + +ARTIFACTS=("build_info.json" "release_notes.txt") +declare -a REFS=() +declare -a HASHES=() + +for artifact in "${ARTIFACTS[@]}"; do + FILE="$ARTIFACTS_DIR/$artifact" + HASH=$(shasum -a 256 "$FILE" | cut -d' ' -f1) + HASHES+=("$HASH") + echo "Uploading: $artifact" + echo " SHA256: $HASH" + + UPLOAD_OUTPUT=$(swarm-prov-upload upload --file "$FILE" --std "CI-ARTIFACT-V1" --usePool 2>&1) || { + echo " Pool not available, falling back to regular stamp purchase..." + UPLOAD_OUTPUT=$(swarm-prov-upload upload --file "$FILE" --std "CI-ARTIFACT-V1" 2>&1) + } + + SWARM_REF=$(echo "$UPLOAD_OUTPUT" | grep -A1 "Swarm Reference Hash:" | tail -1 | tr -d '[:space:]') + + if [ -z "$SWARM_REF" ] || [ ${#SWARM_REF} -lt 64 ]; then + echo "ERROR: Could not extract Swarm reference for $artifact" + echo "Raw output: $UPLOAD_OUTPUT" + exit 1 + fi + + REFS+=("$SWARM_REF") + echo " Reference: $SWARM_REF" + echo +done + +echo "All ${#ARTIFACTS[@]} artifacts archived." +echo + +# --- Step 3: Save receipt manifest --- +echo "--- Step 3: Save archive receipt ---" +RECEIPT_FILE="$SCRIPT_DIR/archive_receipt.json" +TIMESTAMP=$(date -u +"%Y-%m-%dT%H:%M:%SZ") + +cat > "$RECEIPT_FILE" </dev/null | head -1) +if [ -z "$DOWNLOADED_FILE" ]; then + DOWNLOADED_FILE=$(ls "$DOWNLOAD_DIR"/ | head -1) + DOWNLOADED_FILE="$DOWNLOAD_DIR/$DOWNLOADED_FILE" +fi +DOWNLOADED_HASH=$(shasum -a 256 "$DOWNLOADED_FILE" | cut -d' ' -f1) + +echo "Original: $ORIGINAL_HASH" +echo "Downloaded: $DOWNLOADED_HASH" +echo + +if [ "$ORIGINAL_HASH" = "$DOWNLOADED_HASH" ]; then + echo "PASS: Build artifact integrity verified - hashes match." +else + echo "FAIL: Hash mismatch - artifact may have been tampered with!" + exit 1 +fi + +echo +echo "--- Demo complete ---" +echo "Archived ${#ARTIFACTS[@]} build artifacts with CI-ARTIFACT-V1 standard." +echo "Receipt: $RECEIPT_FILE" diff --git a/examples/08-ci-cd-integration/github-action.yml b/examples/08-ci-cd-integration/github-action.yml new file mode 100644 index 0000000..5c84069 --- /dev/null +++ b/examples/08-ci-cd-integration/github-action.yml @@ -0,0 +1,58 @@ +# Sample GitHub Actions workflow for archiving build artifacts to Swarm +# +# This workflow uploads build artifacts to the Swarm network after a +# successful build, creating an immutable provenance record. +# +# Required secrets: +# PROVENANCE_GATEWAY_URL - Gateway URL (optional, uses default if not set) +# +# Usage: +# Copy this file to .github/workflows/swarm-archive.yml in your repository. + +name: Archive Build Artifacts to Swarm + +on: + push: + branches: [main] + release: + types: [published] + +jobs: + build-and-archive: + runs-on: ubuntu-latest + + steps: + - name: Checkout code + uses: actions/checkout@v4 + + - name: Set up Python + uses: actions/setup-python@v5 + with: + python-version: "3.11" + + - name: Install Swarm Provenance CLI + run: pip install swarm-provenance-uploader + + - name: Build project + run: | + echo "Building project..." + # Replace with your actual build commands + mkdir -p dist + echo '{"version": "${{ github.sha }}"}' > dist/build_info.json + + - name: Check gateway health + run: swarm-prov-upload health + + - name: Upload build artifacts to Swarm + run: | + swarm-prov-upload upload \ + --file dist/build_info.json \ + --std "CI-ARTIFACT-V1" \ + --usePool + env: + PROVENANCE_GATEWAY_URL: ${{ secrets.PROVENANCE_GATEWAY_URL }} + + - name: Save Swarm reference + run: | + echo "Artifact archived to Swarm network" + echo "Reference can be used to retrieve the exact build artifact" diff --git a/examples/08-ci-cd-integration/gitlab-ci.yml b/examples/08-ci-cd-integration/gitlab-ci.yml new file mode 100644 index 0000000..ddbf5dd --- /dev/null +++ b/examples/08-ci-cd-integration/gitlab-ci.yml @@ -0,0 +1,42 @@ +# Sample GitLab CI/CD configuration for archiving build artifacts to Swarm +# +# This pipeline uploads build artifacts to the Swarm network after a +# successful build, creating an immutable provenance record. +# +# Required CI/CD variables: +# PROVENANCE_GATEWAY_URL - Gateway URL (optional, uses default if not set) +# +# Usage: +# Copy relevant sections to your .gitlab-ci.yml file. + +stages: + - build + - archive + +build: + stage: build + image: python:3.11 + script: + - echo "Building project..." + # Replace with your actual build commands + - mkdir -p dist + - echo "{\"version\":\"${CI_COMMIT_SHA}\"}" > dist/build_info.json + artifacts: + paths: + - dist/ + +archive-to-swarm: + stage: archive + image: python:3.11 + needs: + - build + script: + - pip install swarm-provenance-uploader + - swarm-prov-upload health + - swarm-prov-upload upload + --file dist/build_info.json + --std "CI-ARTIFACT-V1" + --usePool + only: + - main + - tags diff --git a/examples/08-ci-cd-integration/run_demo.py b/examples/08-ci-cd-integration/run_demo.py new file mode 100644 index 0000000..6555afb --- /dev/null +++ b/examples/08-ci-cd-integration/run_demo.py @@ -0,0 +1,177 @@ +#!/usr/bin/env python3 +""" +CI/CD Integration Demo - Python Version + +Demonstrates archiving build artifacts to Swarm: +1. Upload build artifacts with --std "CI-ARTIFACT-V1" +2. Save receipt manifest with references and hashes +3. Download and verify one artifact + +Usage: + python run_demo.py + python run_demo.py --artifacts build_info.json release_notes.txt +""" + +import argparse +import hashlib +import json +import os +import subprocess +import sys +from datetime import datetime, timezone +from pathlib import Path + +SCRIPT_DIR = Path(__file__).parent + +DEFAULT_ARTIFACTS = [ + "sample_artifacts/build_info.json", + "sample_artifacts/release_notes.txt", +] + + +def sha256_file(path: str) -> str: + """Compute SHA-256 hash of a file.""" + h = hashlib.sha256() + with open(path, "rb") as f: + for chunk in iter(lambda: f.read(8192), b""): + h.update(chunk) + return h.hexdigest() + + +def run_cli(*args) -> subprocess.CompletedProcess: + """Run a swarm-prov-upload CLI command.""" + cmd = ["swarm-prov-upload"] + list(args) + result = subprocess.run(cmd, capture_output=True, text=True) + return result + + +def extract_swarm_ref(output: str) -> str: + """Extract Swarm reference hash from CLI output.""" + lines = output.splitlines() + for i, line in enumerate(lines): + if "Swarm Reference Hash:" in line and i + 1 < len(lines): + ref = lines[i + 1].strip() + if len(ref) >= 64: + return ref + return "" + + +def main(): + parser = argparse.ArgumentParser(description="CI/CD integration demo") + parser.add_argument( + "--artifacts", "-a", + nargs="+", + default=DEFAULT_ARTIFACTS, + help="Artifact files to archive (default: sample build artifacts)", + ) + args = parser.parse_args() + + print("=" * 55) + print(" Swarm Provenance CLI - CI/CD Integration (Python)") + print("=" * 55) + + # --- Step 1: Check health --- + print("\n--- Step 1: Check gateway health ---") + result = run_cli("health") + if result.returncode != 0: + print(f"Gateway not available: {result.stderr or result.stdout}") + sys.exit(1) + print(result.stdout.strip()) + + # --- Step 2: Upload artifacts --- + print('\n--- Step 2: Upload artifacts with --std "CI-ARTIFACT-V1" ---') + receipt = { + "pipeline": "ci-cd-demo", + "timestamp": datetime.now(timezone.utc).isoformat(), + "standard": "CI-ARTIFACT-V1", + "artifacts": [], + } + + for artifact_rel in args.artifacts: + artifact_path = str(SCRIPT_DIR / artifact_rel) + artifact_name = os.path.basename(artifact_rel) + + if not os.path.exists(artifact_path): + print(f"ERROR: File not found: {artifact_path}") + sys.exit(1) + + content_hash = sha256_file(artifact_path) + print(f"\nUploading: {artifact_name}") + print(f" SHA256: {content_hash}") + + result = run_cli("upload", "--file", artifact_path, "--std", "CI-ARTIFACT-V1", "--usePool") + if result.returncode != 0: + print(" Pool not available, falling back to regular stamp purchase...") + result = run_cli("upload", "--file", artifact_path, "--std", "CI-ARTIFACT-V1") + if result.returncode != 0: + print(f" Upload failed: {result.stderr or result.stdout}") + sys.exit(1) + + swarm_ref = extract_swarm_ref(result.stdout) + if not swarm_ref: + print(" Could not extract Swarm reference from output") + sys.exit(1) + + receipt["artifacts"].append({ + "filename": artifact_name, + "reference": swarm_ref, + "content_hash": content_hash, + }) + print(f" Reference: {swarm_ref}") + + print(f"\nAll {len(receipt['artifacts'])} artifacts archived.") + + # --- Step 3: Save receipt --- + print("\n--- Step 3: Save archive receipt ---") + receipt_path = str(SCRIPT_DIR / "archive_receipt.json") + with open(receipt_path, "w") as f: + json.dump(receipt, f, indent=2) + print(f"Receipt saved: {receipt_path}") + + # --- Step 4: Download and verify first artifact --- + first = receipt["artifacts"][0] + print(f"\n--- Step 4: Download and verify {first['filename']} ---") + download_dir = str(SCRIPT_DIR / "downloads") + os.makedirs(download_dir, exist_ok=True) + for f in os.listdir(download_dir): + os.remove(os.path.join(download_dir, f)) + + result = run_cli("download", first["reference"], "--output-dir", download_dir) + if result.returncode != 0: + print(f"Download failed: {result.stderr or result.stdout}") + sys.exit(1) + print(result.stdout.strip()) + + # --- Step 5: Verify integrity --- + print("\n--- Step 5: Verify integrity ---") + downloaded_files = os.listdir(download_dir) + if not downloaded_files: + print("ERROR: No files in download directory") + sys.exit(1) + + data_files = [f for f in downloaded_files if f.endswith(".data")] + if data_files: + downloaded_file = os.path.join(download_dir, data_files[0]) + else: + downloaded_file = os.path.join(download_dir, downloaded_files[0]) + + downloaded_hash = sha256_file(downloaded_file) + original_hash = first["content_hash"] + + print(f"Original: {original_hash}") + print(f"Downloaded: {downloaded_hash}") + + if original_hash == downloaded_hash: + print("\nPASS: Build artifact integrity verified - hashes match.") + else: + print("\nFAIL: Hash mismatch - artifact may have been tampered with!") + sys.exit(1) + + # --- Summary --- + print("\n--- Summary ---") + print(f"Archived {len(receipt['artifacts'])} build artifacts with CI-ARTIFACT-V1 standard.") + print(f"Receipt: {receipt_path}") + + +if __name__ == "__main__": + main() diff --git a/examples/08-ci-cd-integration/sample_artifacts/build_info.json b/examples/08-ci-cd-integration/sample_artifacts/build_info.json new file mode 100644 index 0000000..634ef19 --- /dev/null +++ b/examples/08-ci-cd-integration/sample_artifacts/build_info.json @@ -0,0 +1,26 @@ +{ + "project": "my-web-app", + "version": "2.1.0", + "build_number": 347, + "git_commit": "a1b2c3d4e5f6a7b8c9d0e1f2a3b4c5d6e7f8a9b0", + "git_branch": "main", + "build_timestamp": "2024-01-15T14:30:00Z", + "builder": "github-actions", + "environment": { + "os": "ubuntu-22.04", + "node_version": "20.10.0", + "python_version": "3.11.7" + }, + "tests": { + "total": 342, + "passed": 342, + "failed": 0, + "skipped": 3, + "duration_seconds": 47.2 + }, + "artifacts": [ + "dist/app.js", + "dist/app.css", + "dist/index.html" + ] +} diff --git a/examples/08-ci-cd-integration/sample_artifacts/release_notes.txt b/examples/08-ci-cd-integration/sample_artifacts/release_notes.txt new file mode 100644 index 0000000..25128bf --- /dev/null +++ b/examples/08-ci-cd-integration/sample_artifacts/release_notes.txt @@ -0,0 +1,26 @@ +Release Notes - v2.1.0 +====================== + +Release Date: 2024-01-15 + +New Features: +- Added dark mode support across all pages +- Implemented real-time notification system +- Added CSV export for analytics dashboard + +Bug Fixes: +- Fixed login timeout on slow connections (#423) +- Resolved pagination issue in search results (#431) +- Fixed currency formatting for non-USD locales (#445) + +Performance: +- Reduced initial page load by 35% +- Optimized database queries for user dashboard +- Implemented lazy loading for image galleries + +Breaking Changes: +- None + +Dependencies Updated: +- React 18.2 -> 18.3 +- PostgreSQL driver 3.4 -> 3.5 diff --git a/examples/09-verification/README.md b/examples/09-verification/README.md new file mode 100644 index 0000000..9f06662 --- /dev/null +++ b/examples/09-verification/README.md @@ -0,0 +1,111 @@ +# Example 09: Verification & Integrity + +Demonstrates data verification, tamper detection, and integrity reporting. + +## What This Demonstrates + +- Uploading a document and verifying download integrity via SHA-256 +- Tamper detection: comparing original vs modified file hashes +- Using `--verify` flag for notary signature verification +- Generating a verification report + +## Use Case + +Content-addressed storage provides built-in integrity guarantees: any change to the data produces a different hash. This example shows how to: + +- **Verify downloads**: Confirm that a downloaded file matches the original +- **Detect tampering**: Show that even a small change (e.g., changing "24 months" to "36 months" in a contract) produces a completely different hash +- **Notary verification**: Use the `--verify` flag to check for notary signatures on downloaded data +- **Build trust reports**: Generate verification reports for compliance or audit purposes + +## Prerequisites + +1. Install the CLI: `pip install -e .` +2. Gateway access (default, no setup needed) + +## Quick Start + +### Shell + +```bash +chmod +x demo.sh +./demo.sh +``` + +### Python + +```bash +python run_demo.py +``` + +## Step-by-Step Walkthrough + +### 1. Upload original document + +```bash +swarm-prov-upload upload --file sample_document.txt --usePool +``` + +### 2. Download and verify + +```bash +swarm-prov-upload download --output-dir ./downloads +shasum -a 256 sample_document.txt downloads/*.data +``` + +Hashes must match — proving the document is intact. + +### 3. Tamper detection + +Compare the original and tampered files: + +```bash +shasum -a 256 sample_document.txt sample_document_tampered.txt +``` + +The tampered file has "24 months" changed to "36 months" — a single word change that produces a completely different hash. + +### 4. Notary verification + +```bash +swarm-prov-upload download --output-dir ./downloads --verify +``` + +The `--verify` flag checks for notary signatures. If no signatures exist, the download still succeeds. + +## Helper Tools + +### Integrity Checker + +Verify a Swarm reference against an expected hash: + +```bash +python integrity_checker.py --ref --original-file sample_document.txt +python integrity_checker.py --ref --expected-hash +``` + +### Tamper Detection + +Compare two files and show differences: + +```bash +python tamper_detection.py --original sample_document.txt --tampered sample_document_tampered.txt +``` + +## Sample Documents + +| File | Description | +|------|-------------| +| `sample_document.txt` | Original data processing agreement (24-month retention) | +| `sample_document_tampered.txt` | Tampered version (36-month retention — one word changed) | + +## Files + +| File | Description | +|------|-------------| +| `demo.sh` | Shell demo — upload, verify, tamper test, notary check | +| `run_demo.py` | Python demo — same workflow with argparse support | +| `integrity_checker.py` | Standalone integrity verification tool | +| `tamper_detection.py` | Standalone tamper detection tool | +| `sample_document.txt` | Original document | +| `sample_document_tampered.txt` | Tampered document for comparison | diff --git a/examples/09-verification/demo.sh b/examples/09-verification/demo.sh new file mode 100755 index 0000000..007cf60 --- /dev/null +++ b/examples/09-verification/demo.sh @@ -0,0 +1,130 @@ +#!/usr/bin/env bash +# +# Verification & Integrity Demo +# +# Demonstrates data verification and tamper detection: +# 1. Upload original document +# 2. Download and verify integrity (hash match) +# 3. Tamper test: compare original vs tampered file hashes +# 4. Download with --verify flag (notary verification) +# 5. Print verification report +# +# Usage: ./demo.sh + +set -euo pipefail + +SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)" +ORIGINAL_FILE="$SCRIPT_DIR/sample_document.txt" +TAMPERED_FILE="$SCRIPT_DIR/sample_document_tampered.txt" +DOWNLOAD_DIR="$SCRIPT_DIR/downloads" + +echo "=================================================" +echo " Swarm Provenance CLI - Verification & Integrity" +echo "=================================================" +echo + +# --- Step 0: Check CLI is installed --- +if ! command -v swarm-prov-upload &>/dev/null; then + echo "ERROR: swarm-prov-upload not found. Install with: pip install -e ." + exit 1 +fi + +echo "CLI version: $(swarm-prov-upload --version)" +echo + +# --- Step 1: Check gateway health --- +echo "--- Step 1: Check gateway health ---" +swarm-prov-upload health +echo + +# --- Step 2: Upload original document --- +echo "--- Step 2: Upload original document ---" +ORIGINAL_HASH=$(shasum -a 256 "$ORIGINAL_FILE" | cut -d' ' -f1) +echo "Uploading: sample_document.txt" +echo " SHA256: $ORIGINAL_HASH" + +UPLOAD_OUTPUT=$(swarm-prov-upload upload --file "$ORIGINAL_FILE" --usePool 2>&1) || { + echo " Pool not available, falling back to regular stamp purchase..." + UPLOAD_OUTPUT=$(swarm-prov-upload upload --file "$ORIGINAL_FILE" 2>&1) +} + +SWARM_REF=$(echo "$UPLOAD_OUTPUT" | grep -A1 "Swarm Reference Hash:" | tail -1 | tr -d '[:space:]') + +if [ -z "$SWARM_REF" ] || [ ${#SWARM_REF} -lt 64 ]; then + echo "ERROR: Could not extract Swarm reference" + echo "Raw output: $UPLOAD_OUTPUT" + exit 1 +fi + +echo " Reference: $SWARM_REF" +echo + +# --- Step 3: Download and verify integrity --- +echo "--- Step 3: Download and verify integrity ---" +rm -rf "$DOWNLOAD_DIR" +mkdir -p "$DOWNLOAD_DIR" + +echo "Downloading: $SWARM_REF" +swarm-prov-upload download "$SWARM_REF" --output-dir "$DOWNLOAD_DIR" +echo + +DOWNLOADED_FILE=$(ls "$DOWNLOAD_DIR"/*.data 2>/dev/null | head -1) +if [ -z "$DOWNLOADED_FILE" ]; then + DOWNLOADED_FILE=$(ls "$DOWNLOAD_DIR"/ | head -1) + DOWNLOADED_FILE="$DOWNLOAD_DIR/$DOWNLOADED_FILE" +fi +DOWNLOADED_HASH=$(shasum -a 256 "$DOWNLOADED_FILE" | cut -d' ' -f1) + +echo "Original: $ORIGINAL_HASH" +echo "Downloaded: $DOWNLOADED_HASH" +echo + +if [ "$ORIGINAL_HASH" = "$DOWNLOADED_HASH" ]; then + echo "PASS: Document integrity verified - hashes match." +else + echo "FAIL: Hash mismatch - document may have been tampered with!" + exit 1 +fi +echo + +# --- Step 4: Tamper detection test --- +echo "--- Step 4: Tamper detection test ---" +TAMPERED_HASH=$(shasum -a 256 "$TAMPERED_FILE" | cut -d' ' -f1) +echo "Original document hash: $ORIGINAL_HASH" +echo "Tampered document hash: $TAMPERED_HASH" +echo + +if [ "$ORIGINAL_HASH" != "$TAMPERED_HASH" ]; then + echo "PASS: Tamper detection works - hashes differ." + echo "Even a small change (e.g., '24 months' -> '36 months') produces" + echo "a completely different SHA-256 hash, making tampering detectable." +else + echo "FAIL: Hashes should differ for different content!" + exit 1 +fi +echo + +# --- Step 5: Download with --verify (notary) --- +echo "--- Step 5: Download with --verify (notary verification) ---" +rm -rf "$DOWNLOAD_DIR" +mkdir -p "$DOWNLOAD_DIR" + +echo "Downloading with signature verification..." +VERIFY_OUTPUT=$(swarm-prov-upload download "$SWARM_REF" --output-dir "$DOWNLOAD_DIR" --verify 2>&1) || true +echo "$VERIFY_OUTPUT" +echo +echo "Note: --verify checks for notary signatures. If no signatures exist," +echo "the download still succeeds but reports 'no signatures found'." +echo + +# --- Step 6: Verification report --- +echo "--- Step 6: Verification Report ---" +echo "=================================================" +echo " Document: sample_document.txt" +echo " Swarm Reference: $SWARM_REF" +echo " SHA-256 Hash: $ORIGINAL_HASH" +echo " Integrity: VERIFIED" +echo " Tamper Detection: WORKING" +echo "=================================================" +echo +echo "PASS: All verification checks passed." diff --git a/examples/09-verification/integrity_checker.py b/examples/09-verification/integrity_checker.py new file mode 100644 index 0000000..d2f22c5 --- /dev/null +++ b/examples/09-verification/integrity_checker.py @@ -0,0 +1,133 @@ +#!/usr/bin/env python3 +""" +Integrity Checker + +Verifies a Swarm reference against an expected content hash. +Downloads the file and compares SHA-256 hashes. + +Usage: + python integrity_checker.py --ref --expected-hash + python integrity_checker.py --ref --original-file document.txt +""" + +import argparse +import hashlib +import os +import subprocess +import sys +from pathlib import Path + + +def sha256_file(path: str) -> str: + """Compute SHA-256 hash of a file.""" + h = hashlib.sha256() + with open(path, "rb") as f: + for chunk in iter(lambda: f.read(8192), b""): + h.update(chunk) + return h.hexdigest() + + +def run_cli(*args) -> subprocess.CompletedProcess: + """Run a swarm-prov-upload CLI command.""" + cmd = ["swarm-prov-upload"] + list(args) + result = subprocess.run(cmd, capture_output=True, text=True) + return result + + +def verify_reference(swarm_ref: str, expected_hash: str, + output_dir: str) -> dict: + """Download a Swarm reference and verify against expected hash. + + Returns a verification report dict. + """ + os.makedirs(output_dir, exist_ok=True) + for f in os.listdir(output_dir): + os.remove(os.path.join(output_dir, f)) + + result = run_cli("download", swarm_ref, "--output-dir", output_dir) + if result.returncode != 0: + return { + "status": "ERROR", + "message": f"Download failed: {result.stderr or result.stdout}", + } + + downloaded_files = os.listdir(output_dir) + if not downloaded_files: + return { + "status": "ERROR", + "message": "No files in download directory", + } + + data_files = [f for f in downloaded_files if f.endswith(".data")] + if data_files: + downloaded_file = os.path.join(output_dir, data_files[0]) + else: + downloaded_file = os.path.join(output_dir, downloaded_files[0]) + + actual_hash = sha256_file(downloaded_file) + + return { + "status": "PASS" if actual_hash == expected_hash else "FAIL", + "swarm_reference": swarm_ref, + "expected_hash": expected_hash, + "actual_hash": actual_hash, + "match": actual_hash == expected_hash, + "downloaded_file": downloaded_file, + } + + +def main(): + parser = argparse.ArgumentParser( + description="Verify Swarm reference integrity" + ) + parser.add_argument( + "--ref", "-r", + required=True, + help="Swarm reference hash to verify", + ) + group = parser.add_mutually_exclusive_group(required=True) + group.add_argument( + "--expected-hash", "-e", + help="Expected SHA-256 hash", + ) + group.add_argument( + "--original-file", "-f", + help="Original file to compare against", + ) + parser.add_argument( + "--output-dir", "-o", + default="./verification_downloads", + help="Directory for downloaded files", + ) + args = parser.parse_args() + + if args.original_file: + if not os.path.exists(args.original_file): + print(f"ERROR: File not found: {args.original_file}") + sys.exit(1) + expected_hash = sha256_file(args.original_file) + else: + expected_hash = args.expected_hash + + print(f"Verifying Swarm reference: {args.ref}") + print(f"Expected hash: {expected_hash}") + + report = verify_reference(args.ref, expected_hash, args.output_dir) + + print(f"\nVerification result: {report['status']}") + if report["status"] == "ERROR": + print(f"Error: {report['message']}") + sys.exit(1) + + print(f"Expected: {report['expected_hash']}") + print(f"Downloaded: {report['actual_hash']}") + + if report["match"]: + print("\nPASS: Integrity verified - hashes match.") + else: + print("\nFAIL: Integrity check failed - hashes do not match!") + sys.exit(1) + + +if __name__ == "__main__": + main() diff --git a/examples/09-verification/run_demo.py b/examples/09-verification/run_demo.py new file mode 100644 index 0000000..46bb63e --- /dev/null +++ b/examples/09-verification/run_demo.py @@ -0,0 +1,185 @@ +#!/usr/bin/env python3 +""" +Verification & Integrity Demo - Python Version + +Demonstrates data verification and tamper detection: +1. Upload original document +2. Download and verify integrity (hash match) +3. Tamper test: compare original vs tampered file hashes +4. Download with --verify flag (notary verification) +5. Print verification report + +Usage: + python run_demo.py + python run_demo.py --file sample_document.txt --tampered sample_document_tampered.txt +""" + +import argparse +import hashlib +import os +import subprocess +import sys +from pathlib import Path + +SCRIPT_DIR = Path(__file__).parent + + +def sha256_file(path: str) -> str: + """Compute SHA-256 hash of a file.""" + h = hashlib.sha256() + with open(path, "rb") as f: + for chunk in iter(lambda: f.read(8192), b""): + h.update(chunk) + return h.hexdigest() + + +def run_cli(*args) -> subprocess.CompletedProcess: + """Run a swarm-prov-upload CLI command.""" + cmd = ["swarm-prov-upload"] + list(args) + result = subprocess.run(cmd, capture_output=True, text=True) + return result + + +def extract_swarm_ref(output: str) -> str: + """Extract Swarm reference hash from CLI output.""" + lines = output.splitlines() + for i, line in enumerate(lines): + if "Swarm Reference Hash:" in line and i + 1 < len(lines): + ref = lines[i + 1].strip() + if len(ref) >= 64: + return ref + return "" + + +def main(): + parser = argparse.ArgumentParser(description="Verification & integrity demo") + parser.add_argument( + "--file", "-f", + default="sample_document.txt", + help="Original document to upload and verify (default: sample_document.txt)", + ) + parser.add_argument( + "--tampered", "-t", + default="sample_document_tampered.txt", + help="Tampered document for comparison (default: sample_document_tampered.txt)", + ) + args = parser.parse_args() + + print("=" * 55) + print(" Swarm Provenance CLI - Verification (Python)") + print("=" * 55) + + # --- Step 1: Check health --- + print("\n--- Step 1: Check gateway health ---") + result = run_cli("health") + if result.returncode != 0: + print(f"Gateway not available: {result.stderr or result.stdout}") + sys.exit(1) + print(result.stdout.strip()) + + # Verify files exist + original_path = str(SCRIPT_DIR / args.file) + tampered_path = str(SCRIPT_DIR / args.tampered) + + if not os.path.exists(original_path): + print(f"ERROR: Original file not found: {original_path}") + sys.exit(1) + if not os.path.exists(tampered_path): + print(f"ERROR: Tampered file not found: {tampered_path}") + sys.exit(1) + + # --- Step 2: Upload original document --- + print("\n--- Step 2: Upload original document ---") + original_hash = sha256_file(original_path) + print(f"Uploading: {args.file}") + print(f" SHA256: {original_hash}") + + result = run_cli("upload", "--file", original_path, "--usePool") + if result.returncode != 0: + print(" Pool not available, falling back to regular stamp purchase...") + result = run_cli("upload", "--file", original_path) + if result.returncode != 0: + print(f" Upload failed: {result.stderr or result.stdout}") + sys.exit(1) + + swarm_ref = extract_swarm_ref(result.stdout) + if not swarm_ref: + print(" Could not extract Swarm reference from output") + sys.exit(1) + + print(f" Reference: {swarm_ref}") + + # --- Step 3: Download and verify --- + print("\n--- Step 3: Download and verify integrity ---") + download_dir = str(SCRIPT_DIR / "downloads") + os.makedirs(download_dir, exist_ok=True) + for f in os.listdir(download_dir): + os.remove(os.path.join(download_dir, f)) + + result = run_cli("download", swarm_ref, "--output-dir", download_dir) + if result.returncode != 0: + print(f"Download failed: {result.stderr or result.stdout}") + sys.exit(1) + print(result.stdout.strip()) + + downloaded_files = os.listdir(download_dir) + if not downloaded_files: + print("ERROR: No files in download directory") + sys.exit(1) + + data_files = [f for f in downloaded_files if f.endswith(".data")] + if data_files: + downloaded_file = os.path.join(download_dir, data_files[0]) + else: + downloaded_file = os.path.join(download_dir, downloaded_files[0]) + + downloaded_hash = sha256_file(downloaded_file) + + print(f"Original: {original_hash}") + print(f"Downloaded: {downloaded_hash}") + + if original_hash == downloaded_hash: + print("\nPASS: Document integrity verified - hashes match.") + else: + print("\nFAIL: Hash mismatch - document may have been tampered with!") + sys.exit(1) + + # --- Step 4: Tamper detection test --- + print("\n--- Step 4: Tamper detection test ---") + tampered_hash = sha256_file(tampered_path) + print(f"Original document hash: {original_hash}") + print(f"Tampered document hash: {tampered_hash}") + + if original_hash != tampered_hash: + print("\nPASS: Tamper detection works - hashes differ.") + print("Even a small change produces a completely different SHA-256 hash.") + else: + print("\nFAIL: Hashes should differ for different content!") + sys.exit(1) + + # --- Step 5: Download with --verify --- + print("\n--- Step 5: Download with --verify (notary verification) ---") + for f in os.listdir(download_dir): + os.remove(os.path.join(download_dir, f)) + + result = run_cli("download", swarm_ref, "--output-dir", download_dir, "--verify") + print(result.stdout.strip() if result.stdout else "") + if result.stderr: + print(result.stderr.strip()) + print("Note: --verify checks for notary signatures. If none exist,") + print("the download still succeeds but reports no signatures found.") + + # --- Step 6: Verification report --- + print("\n--- Step 6: Verification Report ---") + print("=" * 55) + print(f" Document: {args.file}") + print(f" Swarm Reference: {swarm_ref}") + print(f" SHA-256 Hash: {original_hash}") + print(f" Integrity: VERIFIED") + print(f" Tamper Detection: WORKING") + print("=" * 55) + print("\nPASS: All verification checks passed.") + + +if __name__ == "__main__": + main() diff --git a/examples/09-verification/sample_document.txt b/examples/09-verification/sample_document.txt new file mode 100644 index 0000000..6fc69aa --- /dev/null +++ b/examples/09-verification/sample_document.txt @@ -0,0 +1,35 @@ +AGREEMENT FOR DATA PROCESSING SERVICES +======================================== + +Agreement ID: DPA-2024-0042 +Date: January 15, 2024 + +PARTIES: + Data Controller: Acme Corporation + Data Processor: DataFlow Services Ltd. + +SCOPE: + This agreement covers the processing of personal data as described + in Schedule A, for the purposes of customer analytics and service + improvement. + +DATA CATEGORIES: + - Customer contact information (name, email, phone) + - Service usage logs (timestamps, features used) + - Aggregated behavioral analytics (anonymized) + +RETENTION PERIOD: + Personal data shall be retained for no longer than 24 months from + the date of collection, after which it must be securely deleted. + +SECURITY MEASURES: + - AES-256 encryption at rest + - TLS 1.3 for data in transit + - Role-based access control + - Annual penetration testing + +This document is stored on Swarm for immutable record-keeping. +Any modification will result in a different content hash, proving +the document has been tampered with. + +Signed: ____________________ Date: ____________________ diff --git a/examples/09-verification/sample_document_tampered.txt b/examples/09-verification/sample_document_tampered.txt new file mode 100644 index 0000000..2c3bca5 --- /dev/null +++ b/examples/09-verification/sample_document_tampered.txt @@ -0,0 +1,35 @@ +AGREEMENT FOR DATA PROCESSING SERVICES +======================================== + +Agreement ID: DPA-2024-0042 +Date: January 15, 2024 + +PARTIES: + Data Controller: Acme Corporation + Data Processor: DataFlow Services Ltd. + +SCOPE: + This agreement covers the processing of personal data as described + in Schedule A, for the purposes of customer analytics and service + improvement. + +DATA CATEGORIES: + - Customer contact information (name, email, phone) + - Service usage logs (timestamps, features used) + - Aggregated behavioral analytics (anonymized) + +RETENTION PERIOD: + Personal data shall be retained for no longer than 36 months from + the date of collection, after which it must be securely deleted. + +SECURITY MEASURES: + - AES-256 encryption at rest + - TLS 1.3 for data in transit + - Role-based access control + - Annual penetration testing + +This document is stored on Swarm for immutable record-keeping. +Any modification will result in a different content hash, proving +the document has been tampered with. + +Signed: ____________________ Date: ____________________ diff --git a/examples/09-verification/tamper_detection.py b/examples/09-verification/tamper_detection.py new file mode 100644 index 0000000..731588a --- /dev/null +++ b/examples/09-verification/tamper_detection.py @@ -0,0 +1,108 @@ +#!/usr/bin/env python3 +""" +Tamper Detection Tool + +Demonstrates how content-addressed storage detects tampering. +Compares original and modified files to show that any change +produces a completely different hash. + +Usage: + python tamper_detection.py --original document.txt --tampered document_tampered.txt +""" + +import argparse +import hashlib +import os +import sys +from pathlib import Path + + +def sha256_file(path: str) -> str: + """Compute SHA-256 hash of a file.""" + h = hashlib.sha256() + with open(path, "rb") as f: + for chunk in iter(lambda: f.read(8192), b""): + h.update(chunk) + return h.hexdigest() + + +def find_differences(original_path: str, tampered_path: str) -> list: + """Find line-level differences between two text files.""" + with open(original_path, "r") as f: + original_lines = f.readlines() + with open(tampered_path, "r") as f: + tampered_lines = f.readlines() + + diffs = [] + max_lines = max(len(original_lines), len(tampered_lines)) + for i in range(max_lines): + orig = original_lines[i] if i < len(original_lines) else "" + tamp = tampered_lines[i] if i < len(tampered_lines) else "" + if orig != tamp: + diffs.append({ + "line": i + 1, + "original": orig.rstrip("\n"), + "tampered": tamp.rstrip("\n"), + }) + return diffs + + +def main(): + parser = argparse.ArgumentParser( + description="Detect tampering via hash comparison" + ) + parser.add_argument( + "--original", "-o", + required=True, + help="Path to original file", + ) + parser.add_argument( + "--tampered", "-t", + required=True, + help="Path to potentially tampered file", + ) + args = parser.parse_args() + + if not os.path.exists(args.original): + print(f"ERROR: Original file not found: {args.original}") + sys.exit(1) + if not os.path.exists(args.tampered): + print(f"ERROR: Tampered file not found: {args.tampered}") + sys.exit(1) + + print("=" * 55) + print(" Tamper Detection Report") + print("=" * 55) + + original_hash = sha256_file(args.original) + tampered_hash = sha256_file(args.tampered) + + print(f"\nOriginal file: {args.original}") + print(f" SHA-256: {original_hash}") + print(f"\nCompared file: {args.tampered}") + print(f" SHA-256: {tampered_hash}") + + if original_hash == tampered_hash: + print("\nRESULT: Files are IDENTICAL (hashes match)") + print("No tampering detected.") + else: + print("\nRESULT: Files are DIFFERENT (hashes do not match)") + print("TAMPERING DETECTED!") + + # Show differences + diffs = find_differences(args.original, args.tampered) + if diffs: + print(f"\nDifferences found ({len(diffs)} lines changed):") + for diff in diffs[:5]: # Show first 5 differences + print(f" Line {diff['line']}:") + print(f" Original: {diff['original']}") + print(f" Tampered: {diff['tampered']}") + if len(diffs) > 5: + print(f" ... and {len(diffs) - 5} more differences") + + return original_hash != tampered_hash + + +if __name__ == "__main__": + tampered = main() + sys.exit(0) diff --git a/examples/README.md b/examples/README.md index dc09fd5..1842b65 100644 --- a/examples/README.md +++ b/examples/README.md @@ -26,8 +26,12 @@ Real-world usage examples for the Swarm Provenance CLI toolkit. | [01](01-basic-upload-download/) | **Basic Upload/Download** | Upload a file, download it back, verify integrity | | [02](02-audit-trail/) | **Audit Trail** | Immutable compliance records with `--std "AUDIT-LOG-V1"` | | [03](03-scientific-data/) | **Scientific Data** | Research archival with `--std "PROV-O"` and `--duration 720` | +| [04](04-batch-processing/) | **Batch Processing** | Stamp reuse across multiple uploads with `--stamp-id` | | [05](05-encrypted-data/) | **Encrypted Data** | Pre-encrypt, upload with `--enc "AES-256-GCM"`, decrypt | | [06](06-market-memory/) | **Market Memory** | Canonical hashing, prediction→outcome linking | +| [07](07-stamp-management/) | **Stamp Management** | Full stamp lifecycle: list, info, check, extend, pool-status | +| [08](08-ci-cd-integration/) | **CI/CD Integration** | Archive build artifacts with GitHub Actions / GitLab CI | +| [09](09-verification/) | **Verification & Integrity** | Tamper detection, SHA-256 verification, `--verify` flag | ## Directory Structure @@ -44,10 +48,18 @@ examples/ README.md, demo.sh, run_demo.py, audit_record_*.json 03-scientific-data/ # Research data archival README.md, demo.sh, run_demo.py, dataset_metadata.json, experiment_results.csv + 04-batch-processing/ # Stamp reuse across multiple uploads + README.md, demo.sh, run_demo.py, batch_upload.py, sample_files/*.json 05-encrypted-data/ # Pre-encryption workflow README.md, demo.sh, run_demo.py, sensitive_data.txt 06-market-memory/ # Prediction memory units README.md, demo.sh, run_demo.py, create_memory_unit.py, prediction_001.json, observation_001.json + 07-stamp-management/ # Full stamp lifecycle + README.md, demo.sh, run_demo.py, stamp_lifecycle.py, sample_data.txt + 08-ci-cd-integration/ # CI/CD artifact archival + README.md, demo.sh, run_demo.py, archive_artifacts.py, github-action.yml, gitlab-ci.yml + 09-verification/ # Tamper detection and integrity + README.md, demo.sh, run_demo.py, integrity_checker.py, tamper_detection.py ``` ## Common Utilities diff --git a/tests/test_examples.py b/tests/test_examples.py index 982957a..2e92d59 100644 --- a/tests/test_examples.py +++ b/tests/test_examples.py @@ -6,8 +6,12 @@ - demo workflow: full upload/download/verify cycle with mocked CLI - 02-audit-trail: multi-record upload with --std AUDIT-LOG-V1 - 03-scientific-data: PROV-O standard with --duration 720 +- 04-batch-processing: stamp reuse across multiple uploads - 05-encrypted-data: encryption workflow with --enc AES-256-GCM - 06-market-memory: canonical hashing and prediction→observation linking +- 07-stamp-management: full stamp lifecycle commands +- 08-ci-cd-integration: CI/CD artifact archival with CI-ARTIFACT-V1 +- 09-verification: tamper detection and integrity verification Integration tests (marked @pytest.mark.gateway) run actual demos against a live gateway — skipped when gateway is unavailable. @@ -30,8 +34,12 @@ DEMO_DIR = EXAMPLES_DIR / "01-basic-upload-download" AUDIT_DIR = EXAMPLES_DIR / "02-audit-trail" SCIENCE_DIR = EXAMPLES_DIR / "03-scientific-data" +BATCH_DIR = EXAMPLES_DIR / "04-batch-processing" ENCRYPTED_DIR = EXAMPLES_DIR / "05-encrypted-data" MARKET_DIR = EXAMPLES_DIR / "06-market-memory" +STAMP_DIR = EXAMPLES_DIR / "07-stamp-management" +CICD_DIR = EXAMPLES_DIR / "08-ci-cd-integration" +VERIFY_DIR = EXAMPLES_DIR / "09-verification" sys.path.insert(0, str(EXAMPLES_DIR)) from common.sample_generator import generate_text_file, generate_json_file, generate_csv_file @@ -245,6 +253,39 @@ def _upload_success_output(ref_hash=FAKE_HASH): ) +FAKE_STAMP_ID = "b" * 64 + + +def _verbose_upload_output(ref_hash=FAKE_HASH, stamp_id=FAKE_STAMP_ID): + """Upload output that includes verbose stamp ID line.""" + return ( + f"Processing file: sample.txt...\n" + f"Acquiring stamp from pool...\n" + f" Stamp ID Received: {stamp_id}\n" + f"Uploading data to Swarm...\n" + f"\n" + f"SUCCESS! Upload complete.\n" + f"Swarm Reference Hash:\n" + f"{ref_hash}\n" + ) + + +def _stamps_list_output(): + return "Stamps:\n ID: abc123... | Depth: 17 | Amount: 10000000 | Usable: true\n" + + +def _stamps_info_output(stamp_id=FAKE_STAMP_ID): + return f"Stamp ID: {stamp_id}\nDepth: 17\nAmount: 10000000\nUsable: true\nTTL: 86400\n" + + +def _stamps_check_output(): + return "Stamp health: OK\n" + + +def _stamps_pool_status_output(): + return "Pool enabled: true\nAvailable stamps: 5\n" + + class TestDemoShellScript: """Tests for the bash demo script.""" @@ -1561,3 +1602,1184 @@ def test_shell_demo_e2e(self): f"Market memory shell demo failed:\nstdout: {result.stdout}\nstderr: {result.stderr}" ) assert "PASS" in result.stdout + + +# ============================================================================= +# EXAMPLE 04: BATCH PROCESSING TESTS +# ============================================================================= + + +class TestBatchProcessingShellScript: + """Tests for the 04-batch-processing bash demo script.""" + + def test_shell_script_is_valid_bash(self): + result = subprocess.run( + ["bash", "-n", str(BATCH_DIR / "demo.sh")], + capture_output=True, text=True, + ) + assert result.returncode == 0, f"Bash syntax error: {result.stderr}" + + def test_shell_script_is_executable(self): + assert os.access(str(BATCH_DIR / "demo.sh"), os.X_OK) + + def test_shell_script_has_shebang(self): + content = (BATCH_DIR / "demo.sh").read_text() + assert content.startswith("#!/usr/bin/env bash") + + def test_shell_script_uses_strict_mode(self): + content = (BATCH_DIR / "demo.sh").read_text() + assert "set -euo pipefail" in content + + def test_shell_script_uses_size_medium(self): + content = (BATCH_DIR / "demo.sh").read_text() + assert "--size medium" in content + + def test_shell_script_uses_stamp_id(self): + content = (BATCH_DIR / "demo.sh").read_text() + assert "--stamp-id" in content + + +class TestBatchProcessingSampleFiles: + """Tests for batch processing sample JSON files.""" + + @pytest.mark.parametrize("filename", [ + "sample_files/log_entry_001.json", + "sample_files/log_entry_002.json", + "sample_files/log_entry_003.json", + ]) + def test_log_entry_exists(self, filename): + assert (BATCH_DIR / filename).exists() + + @pytest.mark.parametrize("filename", [ + "sample_files/log_entry_001.json", + "sample_files/log_entry_002.json", + "sample_files/log_entry_003.json", + ]) + def test_log_entry_is_valid_json(self, filename): + data = json.loads((BATCH_DIR / filename).read_text()) + assert "log_id" in data + assert "timestamp" in data + assert "level" in data + assert "service" in data + assert "event" in data + + def test_log_entries_have_distinct_levels(self): + levels = set() + for filename in ["sample_files/log_entry_001.json", "sample_files/log_entry_002.json", + "sample_files/log_entry_003.json"]: + data = json.loads((BATCH_DIR / filename).read_text()) + levels.add(data["level"]) + assert len(levels) == 3 + + +class TestBatchUploadHelper: + """Tests for batch_upload.py helper functions.""" + + def _load_module(self): + import importlib.util + spec = importlib.util.spec_from_file_location( + "batch_upload", str(BATCH_DIR / "batch_upload.py") + ) + mod = importlib.util.module_from_spec(spec) + spec.loader.exec_module(mod) + return mod + + def test_extract_stamp_id(self): + mod = self._load_module() + output = f" Stamp ID Received: {FAKE_STAMP_ID}\nOther output\n" + assert mod.extract_stamp_id(output) == FAKE_STAMP_ID + + def test_extract_stamp_id_empty(self): + mod = self._load_module() + assert mod.extract_stamp_id("No stamp here\n") == "" + + +class TestBatchProcessingPythonDemo: + """Tests for 04-batch-processing/run_demo.py with mocked CLI.""" + + def _load_demo(self): + import importlib.util + spec = importlib.util.spec_from_file_location( + "batch_demo", str(BATCH_DIR / "run_demo.py") + ) + mod = importlib.util.module_from_spec(spec) + spec.loader.exec_module(mod) + return mod + + def test_full_workflow(self, tmp_path, monkeypatch): + """Test uploading 3 files with stamp reuse, downloading and verifying.""" + sample_dir = tmp_path / "sample_files" + sample_dir.mkdir() + for f in ["log_entry_001.json", "log_entry_002.json", "log_entry_003.json"]: + content = (BATCH_DIR / "sample_files" / f).read_bytes() + (sample_dir / f).write_bytes(content) + + upload_count = {"n": 0} + + def mock_subprocess_run(cmd, **kwargs): + subcmd = _get_cli_subcommand(cmd) + if subcmd == "health": + return _make_completed_process(stdout="Healthy\n") + elif subcmd == "upload": + upload_count["n"] += 1 + cmd_str = " ".join(str(c) for c in cmd) + if "-v" in cmd_str: + return _make_completed_process( + stdout=_verbose_upload_output() + ) + return _make_completed_process(stdout=_upload_success_output()) + elif subcmd == "download": + dl_dir = tmp_path / "downloads" + dl_dir.mkdir(exist_ok=True) + content = (sample_dir / "log_entry_001.json").read_bytes() + (dl_dir / f"{FAKE_HASH}.data").write_bytes(content) + return _make_completed_process(stdout="Downloaded.\n") + return _make_completed_process() + + mod = self._load_demo() + monkeypatch.setattr(mod.subprocess, "run", mock_subprocess_run) + mod.SCRIPT_DIR = tmp_path + monkeypatch.setattr(sys, "argv", [ + "run_demo.py", "--files", + "sample_files/log_entry_001.json", + "sample_files/log_entry_002.json", + "sample_files/log_entry_003.json", + ]) + mod.main() + assert upload_count["n"] == 3 + + def test_pool_fallback(self, tmp_path, monkeypatch): + """Test fallback when pool is unavailable.""" + sample_dir = tmp_path / "sample_files" + sample_dir.mkdir() + content = (BATCH_DIR / "sample_files" / "log_entry_001.json").read_bytes() + (sample_dir / "log_entry_001.json").write_bytes(content) + + upload_calls = {"n": 0} + + def mock_subprocess_run(cmd, **kwargs): + subcmd = _get_cli_subcommand(cmd) + if subcmd == "health": + return _make_completed_process(stdout="Healthy\n") + elif subcmd == "upload": + upload_calls["n"] += 1 + cmd_str = " ".join(str(c) for c in cmd) + if "--usePool" in cmd_str: + return _make_completed_process(returncode=1, stderr="No pool") + return _make_completed_process(stdout=_verbose_upload_output()) + elif subcmd == "download": + dl_dir = tmp_path / "downloads" + dl_dir.mkdir(exist_ok=True) + (dl_dir / f"{FAKE_HASH}.data").write_bytes(content) + return _make_completed_process(stdout="Downloaded.\n") + return _make_completed_process() + + mod = self._load_demo() + monkeypatch.setattr(mod.subprocess, "run", mock_subprocess_run) + mod.SCRIPT_DIR = tmp_path + monkeypatch.setattr(sys, "argv", [ + "run_demo.py", "--files", "sample_files/log_entry_001.json" + ]) + mod.main() + assert upload_calls["n"] == 2 + + def test_upload_failure_exits(self, tmp_path, monkeypatch): + """Test that total upload failure exits with code 1.""" + sample_dir = tmp_path / "sample_files" + sample_dir.mkdir() + (sample_dir / "log_entry_001.json").write_text('{"test": true}') + + def mock_subprocess_run(cmd, **kwargs): + subcmd = _get_cli_subcommand(cmd) + if subcmd == "health": + return _make_completed_process(stdout="Healthy\n") + elif subcmd == "upload": + return _make_completed_process(returncode=1, stderr="Failed") + return _make_completed_process() + + mod = self._load_demo() + monkeypatch.setattr(mod.subprocess, "run", mock_subprocess_run) + mod.SCRIPT_DIR = tmp_path + monkeypatch.setattr(sys, "argv", [ + "run_demo.py", "--files", "sample_files/log_entry_001.json" + ]) + with pytest.raises(SystemExit) as exc_info: + mod.main() + assert exc_info.value.code == 1 + + def test_hash_mismatch_exits(self, tmp_path, monkeypatch): + """Test that hash mismatch on download exits with code 1.""" + sample_dir = tmp_path / "sample_files" + sample_dir.mkdir() + (sample_dir / "log_entry_001.json").write_bytes(b"original content") + + def mock_subprocess_run(cmd, **kwargs): + subcmd = _get_cli_subcommand(cmd) + if subcmd == "health": + return _make_completed_process(stdout="Healthy\n") + elif subcmd == "upload": + return _make_completed_process(stdout=_verbose_upload_output()) + elif subcmd == "download": + dl_dir = tmp_path / "downloads" + dl_dir.mkdir(exist_ok=True) + (dl_dir / f"{FAKE_HASH}.data").write_bytes(b"different content") + return _make_completed_process(stdout="Downloaded.\n") + return _make_completed_process() + + mod = self._load_demo() + monkeypatch.setattr(mod.subprocess, "run", mock_subprocess_run) + mod.SCRIPT_DIR = tmp_path + monkeypatch.setattr(sys, "argv", [ + "run_demo.py", "--files", "sample_files/log_entry_001.json" + ]) + with pytest.raises(SystemExit) as exc_info: + mod.main() + assert exc_info.value.code == 1 + + def test_missing_file_exits(self, tmp_path, monkeypatch): + """Test exit when input file doesn't exist.""" + def mock_subprocess_run(cmd, **kwargs): + subcmd = _get_cli_subcommand(cmd) + if subcmd == "health": + return _make_completed_process(stdout="Healthy\n") + return _make_completed_process() + + mod = self._load_demo() + monkeypatch.setattr(mod.subprocess, "run", mock_subprocess_run) + mod.SCRIPT_DIR = tmp_path + monkeypatch.setattr(sys, "argv", [ + "run_demo.py", "--files", "sample_files/nonexistent.json" + ]) + with pytest.raises(SystemExit) as exc_info: + mod.main() + assert exc_info.value.code == 1 + + +# ============================================================================= +# EXAMPLE 07: STAMP MANAGEMENT TESTS +# ============================================================================= + + +class TestStampManagementShellScript: + """Tests for the 07-stamp-management bash demo script.""" + + def test_shell_script_is_valid_bash(self): + result = subprocess.run( + ["bash", "-n", str(STAMP_DIR / "demo.sh")], + capture_output=True, text=True, + ) + assert result.returncode == 0, f"Bash syntax error: {result.stderr}" + + def test_shell_script_is_executable(self): + assert os.access(str(STAMP_DIR / "demo.sh"), os.X_OK) + + def test_shell_script_has_shebang(self): + content = (STAMP_DIR / "demo.sh").read_text() + assert content.startswith("#!/usr/bin/env bash") + + def test_shell_script_uses_strict_mode(self): + content = (STAMP_DIR / "demo.sh").read_text() + assert "set -euo pipefail" in content + + def test_shell_script_uses_stamps_commands(self): + content = (STAMP_DIR / "demo.sh").read_text() + assert "stamps pool-status" in content + assert "stamps list" in content + assert "stamps info" in content + assert "stamps check" in content + + +class TestStampManagementSampleFiles: + """Tests for stamp management sample files.""" + + def test_sample_data_exists(self): + assert (STAMP_DIR / "sample_data.txt").exists() + + def test_sample_data_has_content(self): + content = (STAMP_DIR / "sample_data.txt").read_text() + assert len(content) > 0 + assert "Stamp" in content + + +class TestStampLifecycleHelper: + """Tests for stamp_lifecycle.py helper functions.""" + + def _load_module(self): + import importlib.util + spec = importlib.util.spec_from_file_location( + "stamp_lifecycle", str(STAMP_DIR / "stamp_lifecycle.py") + ) + mod = importlib.util.module_from_spec(spec) + spec.loader.exec_module(mod) + return mod + + def test_extract_stamp_id(self): + mod = self._load_module() + output = f" Stamp ID Received: {FAKE_STAMP_ID}\nOther output\n" + assert mod.extract_stamp_id(output) == FAKE_STAMP_ID + + def test_extract_stamp_id_empty(self): + mod = self._load_module() + assert mod.extract_stamp_id("No stamp here\n") == "" + + +class TestStampManagementPythonDemo: + """Tests for 07-stamp-management/run_demo.py with mocked CLI.""" + + def _load_demo(self): + import importlib.util + spec = importlib.util.spec_from_file_location( + "stamp_demo", str(STAMP_DIR / "run_demo.py") + ) + mod = importlib.util.module_from_spec(spec) + spec.loader.exec_module(mod) + return mod + + def test_full_workflow(self, tmp_path, monkeypatch): + """Test upload, stamps list, info, check lifecycle.""" + (tmp_path / "sample_data.txt").write_bytes( + (STAMP_DIR / "sample_data.txt").read_bytes() + ) + + def mock_subprocess_run(cmd, **kwargs): + subcmd = _get_cli_subcommand(cmd) + if subcmd == "health": + return _make_completed_process(stdout="Healthy\n") + elif subcmd == "upload": + return _make_completed_process( + stdout=_verbose_upload_output() + ) + elif subcmd == "stamps": + cmd_str = " ".join(str(c) for c in cmd) + if "pool-status" in cmd_str: + return _make_completed_process(stdout=_stamps_pool_status_output()) + elif "list" in cmd_str: + return _make_completed_process(stdout=_stamps_list_output()) + elif "info" in cmd_str: + return _make_completed_process(stdout=_stamps_info_output()) + elif "check" in cmd_str: + return _make_completed_process(stdout=_stamps_check_output()) + return _make_completed_process(stdout="OK\n") + return _make_completed_process() + + mod = self._load_demo() + monkeypatch.setattr(mod.subprocess, "run", mock_subprocess_run) + mod.SCRIPT_DIR = tmp_path + monkeypatch.setattr(sys, "argv", [ + "run_demo.py", "--file", "sample_data.txt" + ]) + mod.main() + + def test_pool_fallback(self, tmp_path, monkeypatch): + """Test fallback when pool is unavailable.""" + (tmp_path / "sample_data.txt").write_bytes( + (STAMP_DIR / "sample_data.txt").read_bytes() + ) + + upload_calls = {"n": 0} + + def mock_subprocess_run(cmd, **kwargs): + subcmd = _get_cli_subcommand(cmd) + if subcmd == "health": + return _make_completed_process(stdout="Healthy\n") + elif subcmd == "upload": + upload_calls["n"] += 1 + cmd_str = " ".join(str(c) for c in cmd) + if "--usePool" in cmd_str: + return _make_completed_process(returncode=1, stderr="No pool") + return _make_completed_process(stdout=_verbose_upload_output()) + elif subcmd == "stamps": + return _make_completed_process(stdout="OK\n") + return _make_completed_process() + + mod = self._load_demo() + monkeypatch.setattr(mod.subprocess, "run", mock_subprocess_run) + mod.SCRIPT_DIR = tmp_path + monkeypatch.setattr(sys, "argv", [ + "run_demo.py", "--file", "sample_data.txt" + ]) + mod.main() + assert upload_calls["n"] == 2 + + def test_upload_failure_exits(self, tmp_path, monkeypatch): + """Test exit when upload fails.""" + (tmp_path / "sample_data.txt").write_text("test data") + + def mock_subprocess_run(cmd, **kwargs): + subcmd = _get_cli_subcommand(cmd) + if subcmd == "health": + return _make_completed_process(stdout="Healthy\n") + elif subcmd == "upload": + return _make_completed_process(returncode=1, stderr="Failed") + elif subcmd == "stamps": + return _make_completed_process(stdout="OK\n") + return _make_completed_process() + + mod = self._load_demo() + monkeypatch.setattr(mod.subprocess, "run", mock_subprocess_run) + mod.SCRIPT_DIR = tmp_path + monkeypatch.setattr(sys, "argv", [ + "run_demo.py", "--file", "sample_data.txt" + ]) + with pytest.raises(SystemExit) as exc_info: + mod.main() + assert exc_info.value.code == 1 + + def test_missing_file_exits(self, tmp_path, monkeypatch): + """Test exit when file doesn't exist.""" + def mock_subprocess_run(cmd, **kwargs): + subcmd = _get_cli_subcommand(cmd) + if subcmd == "health": + return _make_completed_process(stdout="Healthy\n") + elif subcmd == "stamps": + return _make_completed_process(stdout="OK\n") + return _make_completed_process() + + mod = self._load_demo() + monkeypatch.setattr(mod.subprocess, "run", mock_subprocess_run) + mod.SCRIPT_DIR = tmp_path + monkeypatch.setattr(sys, "argv", [ + "run_demo.py", "--file", "nonexistent.txt" + ]) + with pytest.raises(SystemExit) as exc_info: + mod.main() + assert exc_info.value.code == 1 + + def test_no_stamp_id_graceful(self, tmp_path, monkeypatch): + """Test graceful handling when stamp ID cannot be extracted.""" + (tmp_path / "sample_data.txt").write_text("test data") + + def mock_subprocess_run(cmd, **kwargs): + subcmd = _get_cli_subcommand(cmd) + if subcmd == "health": + return _make_completed_process(stdout="Healthy\n") + elif subcmd == "upload": + # No stamp ID in output + return _make_completed_process(stdout=_upload_success_output()) + elif subcmd == "stamps": + return _make_completed_process(stdout="OK\n") + return _make_completed_process() + + mod = self._load_demo() + monkeypatch.setattr(mod.subprocess, "run", mock_subprocess_run) + mod.SCRIPT_DIR = tmp_path + monkeypatch.setattr(sys, "argv", [ + "run_demo.py", "--file", "sample_data.txt" + ]) + # Should not raise — gracefully skips lifecycle steps + mod.main() + + +# ============================================================================= +# EXAMPLE 08: CI/CD INTEGRATION TESTS +# ============================================================================= + + +class TestCiCdShellScript: + """Tests for the 08-ci-cd-integration bash demo script.""" + + def test_shell_script_is_valid_bash(self): + result = subprocess.run( + ["bash", "-n", str(CICD_DIR / "demo.sh")], + capture_output=True, text=True, + ) + assert result.returncode == 0, f"Bash syntax error: {result.stderr}" + + def test_shell_script_is_executable(self): + assert os.access(str(CICD_DIR / "demo.sh"), os.X_OK) + + def test_shell_script_has_shebang(self): + content = (CICD_DIR / "demo.sh").read_text() + assert content.startswith("#!/usr/bin/env bash") + + def test_shell_script_uses_strict_mode(self): + content = (CICD_DIR / "demo.sh").read_text() + assert "set -euo pipefail" in content + + def test_shell_script_uses_ci_artifact_std(self): + content = (CICD_DIR / "demo.sh").read_text() + assert '--std "CI-ARTIFACT-V1"' in content + + +class TestCiCdSampleFiles: + """Tests for CI/CD sample artifacts.""" + + def test_build_info_exists(self): + assert (CICD_DIR / "sample_artifacts" / "build_info.json").exists() + + def test_build_info_is_valid_json(self): + data = json.loads((CICD_DIR / "sample_artifacts" / "build_info.json").read_text()) + assert "project" in data + assert "version" in data + assert "build_number" in data + assert "git_commit" in data + + def test_release_notes_exists(self): + assert (CICD_DIR / "sample_artifacts" / "release_notes.txt").exists() + + def test_release_notes_has_content(self): + content = (CICD_DIR / "sample_artifacts" / "release_notes.txt").read_text() + assert len(content) > 0 + assert "Release" in content + + +class TestCiCdYamlConfigs: + """Tests for CI/CD YAML configuration files.""" + + def test_github_action_exists(self): + assert (CICD_DIR / "github-action.yml").exists() + + def test_github_action_has_swarm_commands(self): + content = (CICD_DIR / "github-action.yml").read_text() + assert "swarm-prov-upload" in content + assert "CI-ARTIFACT-V1" in content + + def test_gitlab_ci_exists(self): + assert (CICD_DIR / "gitlab-ci.yml").exists() + + def test_gitlab_ci_has_swarm_commands(self): + content = (CICD_DIR / "gitlab-ci.yml").read_text() + assert "swarm-prov-upload" in content + assert "CI-ARTIFACT-V1" in content + + +class TestCiCdPythonDemo: + """Tests for 08-ci-cd-integration/run_demo.py with mocked CLI.""" + + def _load_demo(self): + import importlib.util + spec = importlib.util.spec_from_file_location( + "cicd_demo", str(CICD_DIR / "run_demo.py") + ) + mod = importlib.util.module_from_spec(spec) + spec.loader.exec_module(mod) + return mod + + def test_full_workflow(self, tmp_path, monkeypatch): + """Test uploading 2 artifacts, downloading and verifying one.""" + artifacts_dir = tmp_path / "sample_artifacts" + artifacts_dir.mkdir() + for f in ["build_info.json", "release_notes.txt"]: + content = (CICD_DIR / "sample_artifacts" / f).read_bytes() + (artifacts_dir / f).write_bytes(content) + + upload_count = {"n": 0} + + def mock_subprocess_run(cmd, **kwargs): + subcmd = _get_cli_subcommand(cmd) + if subcmd == "health": + return _make_completed_process(stdout="Healthy\n") + elif subcmd == "upload": + upload_count["n"] += 1 + cmd_str = " ".join(str(c) for c in cmd) + assert "CI-ARTIFACT-V1" in cmd_str + return _make_completed_process(stdout=_upload_success_output()) + elif subcmd == "download": + dl_dir = tmp_path / "downloads" + dl_dir.mkdir(exist_ok=True) + content = (artifacts_dir / "build_info.json").read_bytes() + (dl_dir / f"{FAKE_HASH}.data").write_bytes(content) + return _make_completed_process(stdout="Downloaded.\n") + return _make_completed_process() + + mod = self._load_demo() + monkeypatch.setattr(mod.subprocess, "run", mock_subprocess_run) + mod.SCRIPT_DIR = tmp_path + monkeypatch.setattr(sys, "argv", [ + "run_demo.py", "--artifacts", + "sample_artifacts/build_info.json", + "sample_artifacts/release_notes.txt", + ]) + mod.main() + assert upload_count["n"] == 2 + + def test_pool_fallback(self, tmp_path, monkeypatch): + """Test fallback when pool is unavailable.""" + artifacts_dir = tmp_path / "sample_artifacts" + artifacts_dir.mkdir() + content = (CICD_DIR / "sample_artifacts" / "build_info.json").read_bytes() + (artifacts_dir / "build_info.json").write_bytes(content) + + upload_calls = {"n": 0} + + def mock_subprocess_run(cmd, **kwargs): + subcmd = _get_cli_subcommand(cmd) + if subcmd == "health": + return _make_completed_process(stdout="Healthy\n") + elif subcmd == "upload": + upload_calls["n"] += 1 + cmd_str = " ".join(str(c) for c in cmd) + if "--usePool" in cmd_str: + return _make_completed_process(returncode=1, stderr="No pool") + return _make_completed_process(stdout=_upload_success_output()) + elif subcmd == "download": + dl_dir = tmp_path / "downloads" + dl_dir.mkdir(exist_ok=True) + (dl_dir / f"{FAKE_HASH}.data").write_bytes(content) + return _make_completed_process(stdout="Downloaded.\n") + return _make_completed_process() + + mod = self._load_demo() + monkeypatch.setattr(mod.subprocess, "run", mock_subprocess_run) + mod.SCRIPT_DIR = tmp_path + monkeypatch.setattr(sys, "argv", [ + "run_demo.py", "--artifacts", "sample_artifacts/build_info.json" + ]) + mod.main() + assert upload_calls["n"] == 2 + + def test_upload_failure_exits(self, tmp_path, monkeypatch): + """Test that total upload failure exits with code 1.""" + artifacts_dir = tmp_path / "sample_artifacts" + artifacts_dir.mkdir() + (artifacts_dir / "build_info.json").write_text('{"test": true}') + + def mock_subprocess_run(cmd, **kwargs): + subcmd = _get_cli_subcommand(cmd) + if subcmd == "health": + return _make_completed_process(stdout="Healthy\n") + elif subcmd == "upload": + return _make_completed_process(returncode=1, stderr="Failed") + return _make_completed_process() + + mod = self._load_demo() + monkeypatch.setattr(mod.subprocess, "run", mock_subprocess_run) + mod.SCRIPT_DIR = tmp_path + monkeypatch.setattr(sys, "argv", [ + "run_demo.py", "--artifacts", "sample_artifacts/build_info.json" + ]) + with pytest.raises(SystemExit) as exc_info: + mod.main() + assert exc_info.value.code == 1 + + def test_missing_file_exits(self, tmp_path, monkeypatch): + """Test exit when artifact file doesn't exist.""" + mod = self._load_demo() + mod.SCRIPT_DIR = tmp_path + + def mock_subprocess_run(cmd, **kwargs): + subcmd = _get_cli_subcommand(cmd) + if subcmd == "health": + return _make_completed_process(stdout="Healthy\n") + return _make_completed_process() + + monkeypatch.setattr(mod.subprocess, "run", mock_subprocess_run) + monkeypatch.setattr(sys, "argv", [ + "run_demo.py", "--artifacts", "sample_artifacts/nonexistent.json" + ]) + with pytest.raises(SystemExit) as exc_info: + mod.main() + assert exc_info.value.code == 1 + + +# ============================================================================= +# EXAMPLE 09: VERIFICATION & INTEGRITY TESTS +# ============================================================================= + + +class TestVerificationShellScript: + """Tests for the 09-verification bash demo script.""" + + def test_shell_script_is_valid_bash(self): + result = subprocess.run( + ["bash", "-n", str(VERIFY_DIR / "demo.sh")], + capture_output=True, text=True, + ) + assert result.returncode == 0, f"Bash syntax error: {result.stderr}" + + def test_shell_script_is_executable(self): + assert os.access(str(VERIFY_DIR / "demo.sh"), os.X_OK) + + def test_shell_script_has_shebang(self): + content = (VERIFY_DIR / "demo.sh").read_text() + assert content.startswith("#!/usr/bin/env bash") + + def test_shell_script_uses_strict_mode(self): + content = (VERIFY_DIR / "demo.sh").read_text() + assert "set -euo pipefail" in content + + def test_shell_script_uses_verify_flag(self): + content = (VERIFY_DIR / "demo.sh").read_text() + assert "--verify" in content + + +class TestVerificationSampleFiles: + """Tests for verification sample files.""" + + def test_original_exists(self): + assert (VERIFY_DIR / "sample_document.txt").exists() + + def test_tampered_exists(self): + assert (VERIFY_DIR / "sample_document_tampered.txt").exists() + + def test_original_has_content(self): + content = (VERIFY_DIR / "sample_document.txt").read_text() + assert len(content) > 0 + assert "AGREEMENT" in content + + def test_files_are_different(self): + original = (VERIFY_DIR / "sample_document.txt").read_bytes() + tampered = (VERIFY_DIR / "sample_document_tampered.txt").read_bytes() + assert original != tampered + + def test_files_have_different_hashes(self): + orig_hash = hashlib.sha256( + (VERIFY_DIR / "sample_document.txt").read_bytes() + ).hexdigest() + tamp_hash = hashlib.sha256( + (VERIFY_DIR / "sample_document_tampered.txt").read_bytes() + ).hexdigest() + assert orig_hash != tamp_hash + + +class TestTamperDetectionHelper: + """Tests for tamper_detection.py helper.""" + + def _load_module(self): + import importlib.util + spec = importlib.util.spec_from_file_location( + "tamper_detection", str(VERIFY_DIR / "tamper_detection.py") + ) + mod = importlib.util.module_from_spec(spec) + spec.loader.exec_module(mod) + return mod + + def test_find_differences(self): + mod = self._load_module() + diffs = mod.find_differences( + str(VERIFY_DIR / "sample_document.txt"), + str(VERIFY_DIR / "sample_document_tampered.txt"), + ) + assert len(diffs) > 0 + # The tampered file changes "24 months" to "36 months" + changed_text = " ".join(d["tampered"] for d in diffs) + assert "36" in changed_text + + def test_identical_files_no_diffs(self, tmp_path): + mod = self._load_module() + f1 = tmp_path / "same1.txt" + f2 = tmp_path / "same2.txt" + f1.write_text("identical\ncontent\n") + f2.write_text("identical\ncontent\n") + diffs = mod.find_differences(str(f1), str(f2)) + assert len(diffs) == 0 + + +class TestIntegrityCheckerHelper: + """Tests for integrity_checker.py helper.""" + + def _load_module(self): + import importlib.util + spec = importlib.util.spec_from_file_location( + "integrity_checker", str(VERIFY_DIR / "integrity_checker.py") + ) + mod = importlib.util.module_from_spec(spec) + spec.loader.exec_module(mod) + return mod + + def test_verify_reference_match(self, tmp_path, monkeypatch): + """Test successful verification with matching hash.""" + mod = self._load_module() + original_content = b"test document content" + expected_hash = hashlib.sha256(original_content).hexdigest() + output_dir = str(tmp_path / "verify_dl") + + def mock_run(cmd, **kwargs): + subcmd = _get_cli_subcommand(cmd) + if subcmd == "download": + dl_dir = tmp_path / "verify_dl" + dl_dir.mkdir(exist_ok=True) + (dl_dir / f"{FAKE_HASH}.data").write_bytes(original_content) + return _make_completed_process(stdout="Downloaded.\n") + return _make_completed_process() + + monkeypatch.setattr(mod.subprocess, "run", mock_run) + report = mod.verify_reference(FAKE_HASH, expected_hash, output_dir) + assert report["status"] == "PASS" + assert report["match"] is True + + def test_verify_reference_mismatch(self, tmp_path, monkeypatch): + """Test failed verification with mismatched hash.""" + mod = self._load_module() + output_dir = str(tmp_path / "verify_dl") + + def mock_run(cmd, **kwargs): + subcmd = _get_cli_subcommand(cmd) + if subcmd == "download": + dl_dir = tmp_path / "verify_dl" + dl_dir.mkdir(exist_ok=True) + (dl_dir / f"{FAKE_HASH}.data").write_bytes(b"different content") + return _make_completed_process(stdout="Downloaded.\n") + return _make_completed_process() + + monkeypatch.setattr(mod.subprocess, "run", mock_run) + report = mod.verify_reference(FAKE_HASH, "0" * 64, output_dir) + assert report["status"] == "FAIL" + assert report["match"] is False + + +class TestVerificationPythonDemo: + """Tests for 09-verification/run_demo.py with mocked CLI.""" + + def _load_demo(self): + import importlib.util + spec = importlib.util.spec_from_file_location( + "verify_demo", str(VERIFY_DIR / "run_demo.py") + ) + mod = importlib.util.module_from_spec(spec) + spec.loader.exec_module(mod) + return mod + + def test_full_workflow(self, tmp_path, monkeypatch): + """Test upload, verify, tamper detect, --verify workflow.""" + original_content = (VERIFY_DIR / "sample_document.txt").read_bytes() + (tmp_path / "sample_document.txt").write_bytes(original_content) + (tmp_path / "sample_document_tampered.txt").write_bytes( + (VERIFY_DIR / "sample_document_tampered.txt").read_bytes() + ) + + def mock_subprocess_run(cmd, **kwargs): + subcmd = _get_cli_subcommand(cmd) + if subcmd == "health": + return _make_completed_process(stdout="Healthy\n") + elif subcmd == "upload": + return _make_completed_process(stdout=_upload_success_output()) + elif subcmd == "download": + dl_dir = tmp_path / "downloads" + dl_dir.mkdir(exist_ok=True) + # Clear existing files + for f in dl_dir.iterdir(): + f.unlink() + (dl_dir / f"{FAKE_HASH}.data").write_bytes(original_content) + return _make_completed_process(stdout="Downloaded.\n") + return _make_completed_process() + + mod = self._load_demo() + monkeypatch.setattr(mod.subprocess, "run", mock_subprocess_run) + mod.SCRIPT_DIR = tmp_path + monkeypatch.setattr(sys, "argv", [ + "run_demo.py", + "--file", "sample_document.txt", + "--tampered", "sample_document_tampered.txt", + ]) + mod.main() + + def test_pool_fallback(self, tmp_path, monkeypatch): + """Test fallback when pool is unavailable.""" + original_content = (VERIFY_DIR / "sample_document.txt").read_bytes() + (tmp_path / "sample_document.txt").write_bytes(original_content) + (tmp_path / "sample_document_tampered.txt").write_bytes( + (VERIFY_DIR / "sample_document_tampered.txt").read_bytes() + ) + + upload_calls = {"n": 0} + + def mock_subprocess_run(cmd, **kwargs): + subcmd = _get_cli_subcommand(cmd) + if subcmd == "health": + return _make_completed_process(stdout="Healthy\n") + elif subcmd == "upload": + upload_calls["n"] += 1 + cmd_str = " ".join(str(c) for c in cmd) + if "--usePool" in cmd_str: + return _make_completed_process(returncode=1, stderr="No pool") + return _make_completed_process(stdout=_upload_success_output()) + elif subcmd == "download": + dl_dir = tmp_path / "downloads" + dl_dir.mkdir(exist_ok=True) + for f in dl_dir.iterdir(): + f.unlink() + (dl_dir / f"{FAKE_HASH}.data").write_bytes(original_content) + return _make_completed_process(stdout="Downloaded.\n") + return _make_completed_process() + + mod = self._load_demo() + monkeypatch.setattr(mod.subprocess, "run", mock_subprocess_run) + mod.SCRIPT_DIR = tmp_path + monkeypatch.setattr(sys, "argv", [ + "run_demo.py", + "--file", "sample_document.txt", + "--tampered", "sample_document_tampered.txt", + ]) + mod.main() + assert upload_calls["n"] == 2 + + def test_upload_failure_exits(self, tmp_path, monkeypatch): + """Test exit when upload fails.""" + (tmp_path / "sample_document.txt").write_text("test data") + (tmp_path / "sample_document_tampered.txt").write_text("tampered") + + def mock_subprocess_run(cmd, **kwargs): + subcmd = _get_cli_subcommand(cmd) + if subcmd == "health": + return _make_completed_process(stdout="Healthy\n") + elif subcmd == "upload": + return _make_completed_process(returncode=1, stderr="Failed") + return _make_completed_process() + + mod = self._load_demo() + monkeypatch.setattr(mod.subprocess, "run", mock_subprocess_run) + mod.SCRIPT_DIR = tmp_path + monkeypatch.setattr(sys, "argv", [ + "run_demo.py", + "--file", "sample_document.txt", + "--tampered", "sample_document_tampered.txt", + ]) + with pytest.raises(SystemExit) as exc_info: + mod.main() + assert exc_info.value.code == 1 + + def test_hash_mismatch_exits(self, tmp_path, monkeypatch): + """Test exit when downloaded file doesn't match original.""" + (tmp_path / "sample_document.txt").write_bytes(b"original content") + (tmp_path / "sample_document_tampered.txt").write_bytes(b"tampered content") + + def mock_subprocess_run(cmd, **kwargs): + subcmd = _get_cli_subcommand(cmd) + if subcmd == "health": + return _make_completed_process(stdout="Healthy\n") + elif subcmd == "upload": + return _make_completed_process(stdout=_upload_success_output()) + elif subcmd == "download": + dl_dir = tmp_path / "downloads" + dl_dir.mkdir(exist_ok=True) + for f in dl_dir.iterdir(): + f.unlink() + (dl_dir / f"{FAKE_HASH}.data").write_bytes(b"corrupted data") + return _make_completed_process(stdout="Downloaded.\n") + return _make_completed_process() + + mod = self._load_demo() + monkeypatch.setattr(mod.subprocess, "run", mock_subprocess_run) + mod.SCRIPT_DIR = tmp_path + monkeypatch.setattr(sys, "argv", [ + "run_demo.py", + "--file", "sample_document.txt", + "--tampered", "sample_document_tampered.txt", + ]) + with pytest.raises(SystemExit) as exc_info: + mod.main() + assert exc_info.value.code == 1 + + def test_missing_file_exits(self, tmp_path, monkeypatch): + """Test exit when original file doesn't exist.""" + mod = self._load_demo() + mod.SCRIPT_DIR = tmp_path + monkeypatch.setattr(sys, "argv", [ + "run_demo.py", + "--file", "nonexistent.txt", + "--tampered", "also_missing.txt", + ]) + + def mock_subprocess_run(cmd, **kwargs): + subcmd = _get_cli_subcommand(cmd) + if subcmd == "health": + return _make_completed_process(stdout="Healthy\n") + return _make_completed_process() + + monkeypatch.setattr(mod.subprocess, "run", mock_subprocess_run) + with pytest.raises(SystemExit) as exc_info: + mod.main() + assert exc_info.value.code == 1 + + def test_tampered_identical_fails(self, tmp_path, monkeypatch): + """Test that identical original and tampered files cause exit.""" + content = b"same content for both" + (tmp_path / "sample_document.txt").write_bytes(content) + (tmp_path / "sample_document_tampered.txt").write_bytes(content) + + def mock_subprocess_run(cmd, **kwargs): + subcmd = _get_cli_subcommand(cmd) + if subcmd == "health": + return _make_completed_process(stdout="Healthy\n") + elif subcmd == "upload": + return _make_completed_process(stdout=_upload_success_output()) + elif subcmd == "download": + dl_dir = tmp_path / "downloads" + dl_dir.mkdir(exist_ok=True) + for f in dl_dir.iterdir(): + f.unlink() + (dl_dir / f"{FAKE_HASH}.data").write_bytes(content) + return _make_completed_process(stdout="Downloaded.\n") + return _make_completed_process() + + mod = self._load_demo() + monkeypatch.setattr(mod.subprocess, "run", mock_subprocess_run) + mod.SCRIPT_DIR = tmp_path + monkeypatch.setattr(sys, "argv", [ + "run_demo.py", + "--file", "sample_document.txt", + "--tampered", "sample_document_tampered.txt", + ]) + with pytest.raises(SystemExit) as exc_info: + mod.main() + assert exc_info.value.code == 1 + + +# ============================================================================= +# INTEGRATION TESTS — new examples +# ============================================================================= + + +@pytest.mark.integration +@pytest.mark.gateway +class TestBatchProcessingIntegration: + """Integration tests for 04-batch-processing demos.""" + + @skip_if_no_gateway_upload + def test_python_demo_e2e(self): + cli_path = _venv_cli_path() + venv_python = str(Path(__file__).parent.parent / ".venv" / "bin" / "python3") + env = os.environ.copy() + env["PATH"] = str(Path(cli_path).parent) + ":" + env.get("PATH", "") + + result = subprocess.run( + [venv_python, str(BATCH_DIR / "run_demo.py")], + capture_output=True, text=True, + timeout=300, + env=env, + ) + assert result.returncode == 0, ( + f"Batch processing demo failed:\nstdout: {result.stdout}\nstderr: {result.stderr}" + ) + assert "PASS" in result.stdout + + @skip_if_no_gateway_upload + def test_shell_demo_e2e(self): + cli_path = _venv_cli_path() + env = os.environ.copy() + env["PATH"] = str(Path(cli_path).parent) + ":" + env.get("PATH", "") + + result = subprocess.run( + ["bash", str(BATCH_DIR / "demo.sh")], + capture_output=True, text=True, + timeout=300, + env=env, + ) + assert result.returncode == 0, ( + f"Batch processing shell demo failed:\nstdout: {result.stdout}\nstderr: {result.stderr}" + ) + assert "PASS" in result.stdout + + +@pytest.mark.integration +@pytest.mark.gateway +class TestStampManagementIntegration: + """Integration tests for 07-stamp-management demos.""" + + @skip_if_no_gateway_upload + def test_python_demo_e2e(self): + cli_path = _venv_cli_path() + venv_python = str(Path(__file__).parent.parent / ".venv" / "bin" / "python3") + env = os.environ.copy() + env["PATH"] = str(Path(cli_path).parent) + ":" + env.get("PATH", "") + + result = subprocess.run( + [venv_python, str(STAMP_DIR / "run_demo.py")], + capture_output=True, text=True, + timeout=300, + env=env, + ) + assert result.returncode == 0, ( + f"Stamp management demo failed:\nstdout: {result.stdout}\nstderr: {result.stderr}" + ) + assert "PASS" in result.stdout + + @skip_if_no_gateway_upload + def test_shell_demo_e2e(self): + cli_path = _venv_cli_path() + env = os.environ.copy() + env["PATH"] = str(Path(cli_path).parent) + ":" + env.get("PATH", "") + + result = subprocess.run( + ["bash", str(STAMP_DIR / "demo.sh")], + capture_output=True, text=True, + timeout=300, + env=env, + ) + assert result.returncode == 0, ( + f"Stamp management shell demo failed:\nstdout: {result.stdout}\nstderr: {result.stderr}" + ) + assert "PASS" in result.stdout + + +@pytest.mark.integration +@pytest.mark.gateway +class TestCiCdIntegration: + """Integration tests for 08-ci-cd-integration demos.""" + + @skip_if_no_gateway_upload + def test_python_demo_e2e(self): + cli_path = _venv_cli_path() + venv_python = str(Path(__file__).parent.parent / ".venv" / "bin" / "python3") + env = os.environ.copy() + env["PATH"] = str(Path(cli_path).parent) + ":" + env.get("PATH", "") + + result = subprocess.run( + [venv_python, str(CICD_DIR / "run_demo.py")], + capture_output=True, text=True, + timeout=300, + env=env, + ) + assert result.returncode == 0, ( + f"CI/CD demo failed:\nstdout: {result.stdout}\nstderr: {result.stderr}" + ) + assert "PASS" in result.stdout + + @skip_if_no_gateway_upload + def test_shell_demo_e2e(self): + cli_path = _venv_cli_path() + env = os.environ.copy() + env["PATH"] = str(Path(cli_path).parent) + ":" + env.get("PATH", "") + + result = subprocess.run( + ["bash", str(CICD_DIR / "demo.sh")], + capture_output=True, text=True, + timeout=300, + env=env, + ) + assert result.returncode == 0, ( + f"CI/CD shell demo failed:\nstdout: {result.stdout}\nstderr: {result.stderr}" + ) + assert "PASS" in result.stdout + + +@pytest.mark.integration +@pytest.mark.gateway +class TestVerificationIntegration: + """Integration tests for 09-verification demos.""" + + @skip_if_no_gateway_upload + def test_python_demo_e2e(self): + cli_path = _venv_cli_path() + venv_python = str(Path(__file__).parent.parent / ".venv" / "bin" / "python3") + env = os.environ.copy() + env["PATH"] = str(Path(cli_path).parent) + ":" + env.get("PATH", "") + + result = subprocess.run( + [venv_python, str(VERIFY_DIR / "run_demo.py")], + capture_output=True, text=True, + timeout=300, + env=env, + ) + assert result.returncode == 0, ( + f"Verification demo failed:\nstdout: {result.stdout}\nstderr: {result.stderr}" + ) + assert "PASS" in result.stdout + + @skip_if_no_gateway_upload + def test_shell_demo_e2e(self): + cli_path = _venv_cli_path() + env = os.environ.copy() + env["PATH"] = str(Path(cli_path).parent) + ":" + env.get("PATH", "") + + result = subprocess.run( + ["bash", str(VERIFY_DIR / "demo.sh")], + capture_output=True, text=True, + timeout=300, + env=env, + ) + assert result.returncode == 0, ( + f"Verification shell demo failed:\nstdout: {result.stdout}\nstderr: {result.stderr}" + ) + assert "PASS" in result.stdout From 0f322ea8b221082158f617ad5cb012452c0539df Mon Sep 17 00:00:00 2001 From: Crt Ahlin Date: Wed, 25 Feb 2026 21:52:10 +0100 Subject: [PATCH 2/2] Fix stamp ID extraction to handle verbose output format The CLI outputs 'Stamp ID Received: (Length: 64)' in verbose mode. extract_stamp_id() was capturing the full string including the suffix, causing stamp reuse and stamp info/check to fail on the gateway. Fix: extract only the first token (hex ID) after the colon. Update test helper to match real CLI output format. --- examples/04-batch-processing/batch_upload.py | 5 +++-- examples/04-batch-processing/demo.sh | 2 +- examples/04-batch-processing/run_demo.py | 8 ++++++-- examples/07-stamp-management/demo.sh | 2 +- examples/07-stamp-management/run_demo.py | 8 ++++++-- examples/07-stamp-management/stamp_lifecycle.py | 8 ++++++-- tests/test_examples.py | 7 +++++-- 7 files changed, 28 insertions(+), 12 deletions(-) diff --git a/examples/04-batch-processing/batch_upload.py b/examples/04-batch-processing/batch_upload.py index 1355b71..a64068b 100644 --- a/examples/04-batch-processing/batch_upload.py +++ b/examples/04-batch-processing/batch_upload.py @@ -49,13 +49,14 @@ def extract_swarm_ref(output: str) -> str: def extract_stamp_id(output: str) -> str: """Extract stamp ID from verbose CLI output. - Looks for 'Stamp ID Received: ' in verbose output. + Handles format: 'Stamp ID Received: (Length: 64)' """ for line in output.splitlines(): if "Stamp ID Received:" in line: parts = line.split("Stamp ID Received:") if len(parts) > 1: - stamp_id = parts[1].strip() + # Take first token only (ignore trailing "(Length: 64)" etc.) + stamp_id = parts[1].strip().split()[0] if len(stamp_id) >= 16: return stamp_id return "" diff --git a/examples/04-batch-processing/demo.sh b/examples/04-batch-processing/demo.sh index 32528aa..5d6e400 100755 --- a/examples/04-batch-processing/demo.sh +++ b/examples/04-batch-processing/demo.sh @@ -54,7 +54,7 @@ if [ -z "$FIRST_REF" ] || [ ${#FIRST_REF} -lt 64 ]; then fi # Extract stamp ID from verbose output -STAMP_ID=$(echo "$UPLOAD_OUTPUT" | grep "Stamp ID Received:" | awk -F'Stamp ID Received: ' '{print $2}' | tr -d '[:space:]') +STAMP_ID=$(echo "$UPLOAD_OUTPUT" | grep "Stamp ID Received:" | awk -F'Stamp ID Received: ' '{print $2}' | awk '{print $1}' | tr -d '[:space:]') if [ -z "$STAMP_ID" ] || [ ${#STAMP_ID} -lt 16 ]; then echo "WARNING: Could not extract stamp ID from verbose output" echo "Subsequent uploads will purchase new stamps" diff --git a/examples/04-batch-processing/run_demo.py b/examples/04-batch-processing/run_demo.py index 4c16d98..86643d5 100644 --- a/examples/04-batch-processing/run_demo.py +++ b/examples/04-batch-processing/run_demo.py @@ -58,12 +58,16 @@ def extract_swarm_ref(output: str) -> str: def extract_stamp_id(output: str) -> str: - """Extract stamp ID from verbose CLI output.""" + """Extract stamp ID from verbose CLI output. + + Handles format: 'Stamp ID Received: (Length: 64)' + """ for line in output.splitlines(): if "Stamp ID Received:" in line: parts = line.split("Stamp ID Received:") if len(parts) > 1: - stamp_id = parts[1].strip() + # Take first token only (ignore trailing "(Length: 64)" etc.) + stamp_id = parts[1].strip().split()[0] if len(stamp_id) >= 16: return stamp_id return "" diff --git a/examples/07-stamp-management/demo.sh b/examples/07-stamp-management/demo.sh index 6677720..fc5c0af 100755 --- a/examples/07-stamp-management/demo.sh +++ b/examples/07-stamp-management/demo.sh @@ -56,7 +56,7 @@ if [ -z "$SWARM_REF" ] || [ ${#SWARM_REF} -lt 64 ]; then exit 1 fi -STAMP_ID=$(echo "$UPLOAD_OUTPUT" | grep "Stamp ID Received:" | awk -F'Stamp ID Received: ' '{print $2}' | tr -d '[:space:]') +STAMP_ID=$(echo "$UPLOAD_OUTPUT" | grep "Stamp ID Received:" | awk -F'Stamp ID Received: ' '{print $2}' | awk '{print $1}' | tr -d '[:space:]') echo " Swarm reference: $SWARM_REF" if [ -z "$STAMP_ID" ] || [ ${#STAMP_ID} -lt 16 ]; then diff --git a/examples/07-stamp-management/run_demo.py b/examples/07-stamp-management/run_demo.py index 3742897..abfea15 100644 --- a/examples/07-stamp-management/run_demo.py +++ b/examples/07-stamp-management/run_demo.py @@ -52,12 +52,16 @@ def extract_swarm_ref(output: str) -> str: def extract_stamp_id(output: str) -> str: - """Extract stamp ID from verbose CLI output.""" + """Extract stamp ID from verbose CLI output. + + Handles format: 'Stamp ID Received: (Length: 64)' + """ for line in output.splitlines(): if "Stamp ID Received:" in line: parts = line.split("Stamp ID Received:") if len(parts) > 1: - stamp_id = parts[1].strip() + # Take first token only (ignore trailing "(Length: 64)" etc.) + stamp_id = parts[1].strip().split()[0] if len(stamp_id) >= 16: return stamp_id return "" diff --git a/examples/07-stamp-management/stamp_lifecycle.py b/examples/07-stamp-management/stamp_lifecycle.py index 6614f43..0125859 100644 --- a/examples/07-stamp-management/stamp_lifecycle.py +++ b/examples/07-stamp-management/stamp_lifecycle.py @@ -31,12 +31,16 @@ def run_cli(*args) -> subprocess.CompletedProcess: def extract_stamp_id(output: str) -> str: - """Extract stamp ID from verbose CLI output.""" + """Extract stamp ID from verbose CLI output. + + Handles format: 'Stamp ID Received: (Length: 64)' + """ for line in output.splitlines(): if "Stamp ID Received:" in line: parts = line.split("Stamp ID Received:") if len(parts) > 1: - stamp_id = parts[1].strip() + # Take first token only (ignore trailing "(Length: 64)" etc.) + stamp_id = parts[1].strip().split()[0] if len(stamp_id) >= 16: return stamp_id return "" diff --git a/tests/test_examples.py b/tests/test_examples.py index 2e92d59..d33d9db 100644 --- a/tests/test_examples.py +++ b/tests/test_examples.py @@ -257,11 +257,14 @@ def _upload_success_output(ref_hash=FAKE_HASH): def _verbose_upload_output(ref_hash=FAKE_HASH, stamp_id=FAKE_STAMP_ID): - """Upload output that includes verbose stamp ID line.""" + """Upload output that includes verbose stamp ID line. + + Matches real CLI format: 'Stamp ID Received: (Length: 64)' + """ return ( f"Processing file: sample.txt...\n" f"Acquiring stamp from pool...\n" - f" Stamp ID Received: {stamp_id}\n" + f" Stamp ID Received: {stamp_id} (Length: {len(stamp_id)})\n" f"Uploading data to Swarm...\n" f"\n" f"SUCCESS! Upload complete.\n"