A Practical Guide to Retraction: For Makers

import os
import sys
import datetime
from typing import Dict, Any
from flask import Flask, request, jsonify, Response
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor, ConsoleSpanExporter
from opentelemetry.instrumentation.flask import FlaskInstrumentor
import requests

# Configure OpenTelemetry for audit trails
provider = TracerProvider()
processor = BatchSpanProcessor(ConsoleSpanExporter())
provider.add_span_processor(processor)
trace.set_tracer_provider(provider)
tracer = trace.get_tracer(__name__)
FlaskInstrumentor().instrument()

app = Flask(__name__)

# Configuration: load from env vars with defaults
DEPRECATION_STATUS_PAGE_URL = os.getenv("DEPRECATION_STATUS_PAGE_URL", "https://status.example.com/deprecations")
API_VERSION_HEADER = os.getenv("API_VERSION_HEADER", "X-API-Version")
DEPRECATION_NOTICE_HEADER = "X-API-Deprecation-Notice"
DEPRECATION_SUNSET_HEADER = "X-API-Sunset-Date"

# Sample deprecated endpoints mapping: {endpoint_pattern: {sunset_date, replacement}}
DEPRECATED_ENDPOINTS: Dict[str, Dict[str, Any]] = {
    "/v1/products": {
        "sunset_date": "2024-12-31",
        "replacement": "/v2/products",
        "reason": "Legacy pagination logic causes high latency for large catalogs"
    },
    "/v1/checkout": {
        "sunset_date": "2024-11-15",
        "replacement": "/v2/checkout",
        "reason": "Missing support for digital wallet payments"
    }
}

def is_deprecated_endpoint(path: str) -> bool:
    """Check if the request path matches a deprecated endpoint pattern."""
    return any(path.startswith(ep) for ep in DEPRECATED_ENDPOINTS.keys())

def get_deprecation_details(path: str) -> Dict[str, Any]:
    """Retrieve deprecation metadata for a matched endpoint."""
    for ep, details in DEPRECATED_ENDPOINTS.items():
        if path.startswith(ep):
            return details
    return {}

@app.before_request
def inject_deprecation_headers():
    """Middleware to inject deprecation headers and log audit trails."""
    if not is_deprecated_endpoint(request.path):
        return  # Skip non-deprecated endpoints

    # Extract deprecation details
    dep_details = get_deprecation_details(request.path)
    if not dep_details:
        return

    # Start OpenTelemetry span for audit trail
    with tracer.start_as_current_span("api_deprecation_access") as span:
        span.set_attribute("endpoint.path", request.path)
        span.set_attribute("endpoint.method", request.method)
        span.set_attribute("deprecation.sunset_date", dep_details["sunset_date"])
        span.set_attribute("deprecation.replacement", dep_details["replacement"])
        span.set_attribute("client.id", request.headers.get("X-Client-ID", "unknown"))

        # Inject deprecation headers into response (we'll attach later)
        request.deprecation_details = dep_details

        # Log deprecation access to status page (fire and forget)
        try:
            requests.post(
                f"{DEPRECATION_STATUS_PAGE_URL}/log-access",
                json={
                    "endpoint": request.path,
                    "client_id": request.headers.get("X-Client-ID", "unknown"),
                    "timestamp": datetime.datetime.utcnow().isoformat()
                },
                timeout=2  # Short timeout to avoid blocking request
            )
        except requests.exceptions.RequestException as e:
            # Log but don't fail the request if status page is down
            app.logger.warning(f"Failed to log deprecation access to status page: {str(e)}")
            span.set_attribute("status_page.log_error", str(e))

@app.after_request
def add_deprecation_response_headers(response: Response) -> Response:
    """Attach deprecation headers to responses for deprecated endpoints."""
    if not hasattr(request, "deprecation_details"):
        return response

    dep_details = request.deprecation_details
    response.headers[DEPRECATION_NOTICE_HEADER] = (
        f"This endpoint is deprecated. Sunset date: {dep_details['sunset_date']}. "
        f"Replace with: {dep_details['replacement']}. Reason: {dep_details['reason']}"
    )
    response.headers[DEPRECATION_SUNSET_HEADER] = dep_details["sunset_date"]
    response.headers["Link"] = f"<{dep_details['replacement']}>; rel="successor-version""
    return response

@app.route("/v1/products", methods=["GET"])
def get_products_v1():
    """Legacy v1 products endpoint (deprecated)."""
    return jsonify({"products": [], "version": "v1"})

@app.route("/health", methods=["GET"])
def health_check():
    """Health check endpoint (not deprecated)."""
    return jsonify({"status": "healthy"})

if __name__ == "__main__":
    # Validate configuration on startup
    if not os.getenv("DEPRECATION_STATUS_PAGE_URL"):
        app.logger.warning("DEPRECATION_STATUS_PAGE_URL not set, status page logging disabled")
    app.run(host="0.0.0.0", port=8080, debug=False)

import os
import sys
import json
import datetime
import subprocess
from typing import List, Dict, Any, Optional

# Configuration: load from environment variables
GITHUB_REPO = os.getenv("GITHUB_REPO")  # Format: owner/repo
KUBE_NAMESPACE = os.getenv("KUBE_NAMESPACE", "production")
KUBE_DEPLOYMENT = os.getenv("KUBE_DEPLOYMENT", "api-server")
SLACK_WEBHOOK_URL = os.getenv("SLACK_WEBHOOK_URL")
AUDIT_LOG_FILE = os.getenv("AUDIT_LOG_FILE", "/var/log/retraction/audit.log")

def log_audit(event_type: str, details: Dict[str, Any]) -> None:
    """Write structured audit logs to file and OpenTelemetry (simplified here)."""
    log_entry = {
        "timestamp": datetime.datetime.utcnow().isoformat(),
        "event_type": event_type,
        "details": details,
        "actor": os.getenv("ACTOR", "ci-pipeline")
    }
    try:
        with open(AUDIT_LOG_FILE, "a") as f:
            f.write(json.dumps(log_entry) + "\n")
    except IOError as e:
        print(f"CRITICAL: Failed to write audit log: {str(e)}", file=sys.stderr)
        sys.exit(1)

def run_subprocess_command(cmd: List[str], error_msg: str) -> subprocess.CompletedProcess:
    """Run a subprocess command with error handling."""
    try:
        result = subprocess.run(
            cmd,
            capture_output=True,
            text=True,
            check=True
        )
        return result
    except subprocess.CalledProcessError as e:
        log_audit("command_failed", {
            "command": " ".join(cmd),
            "return_code": e.returncode,
            "stdout": e.stdout,
            "stderr": e.stderr
        })
        raise RuntimeError(f"{error_msg}: {e.stderr}") from e

def retract_github_release(release_tag: str) -> None:
    """Retract a GitHub release by marking it as pre-release and adding a retraction notice."""
    if not GITHUB_REPO:
        raise ValueError("GITHUB_REPO environment variable must be set (format: owner/repo)")

    # Fetch release ID from tag
    cmd = ["gh", "release", "view", release_tag, "--repo", GITHUB_REPO, "--json", "id", "--jq", ".id"]
    result = run_subprocess_command(cmd, f"Failed to fetch release ID for tag {release_tag}")

    release_id = result.stdout.strip()
    if not release_id:
        raise ValueError(f"Release with tag {release_tag} not found in {GITHUB_REPO}")

    # Update release to mark as pre-release and add retraction notice
    retraction_note = (
        f"**RETRACTED RELEASE** (retracted on {datetime.datetime.utcnow().strftime('%Y-%m-%d %H:%M UTC')}): "
        f"This release contains a critical bug. Do not use. Roll back to the previous stable release."
    )
    cmd = [
        "gh", "release", "edit", release_tag,
        "--repo", GITHUB_REPO,
        "--prerelease",
        "--notes", retraction_note
    ]
    run_subprocess_command(cmd, f"Failed to retract GitHub release {release_tag}")

    # Delete release artifacts to prevent accidental download
    cmd = ["gh", "release", "delete-asset", release_tag, "--repo", GITHUB_REPO, "--all", "--yes"]
    run_subprocess_command(cmd, f"Failed to delete assets for release {release_tag}")

    log_audit("github_release_retracted", {
        "release_tag": release_tag,
        "release_id": release_id,
        "repo": GITHUB_REPO
    })

def rollback_kubernetes_deployment(target_revision: str) -> None:
    """Roll back Kubernetes deployment to a specific revision."""
    cmd = [
        "kubectl", "rollout", "undo",
        "deployment", KUBE_DEPLOYMENT,
        "-n", KUBE_NAMESPACE,
        "--to-revision", target_revision
    ]
    run_subprocess_command(cmd, f"Failed to roll back deployment {KUBE_DEPLOYMENT} to revision {target_revision}")

    # Wait for rollout to complete
    cmd = ["kubectl", "rollout", "status", "deployment", KUBE_DEPLOYMENT, "-n", KUBE_NAMESPACE]
    run_subprocess_command(cmd, f"Rollout of {KUBE_DEPLOYMENT} failed to stabilize")

    log_audit("kubernetes_rollback", {
        "deployment": KUBE_DEPLOYMENT,
        "namespace": KUBE_NAMESPACE,
        "target_revision": target_revision
    })

def notify_stakeholders(release_tag: str, previous_tag: str) -> None:
    """Send Slack notification about the retraction."""
    if not SLACK_WEBHOOK_URL:
        print("SLACK_WEBHOOK_URL not set, skipping notification")
        return

    message = {
        "text": f"🚨 Release Retraction Alert 🚨",
        "blocks": [
            {
                "type": "section",
                "text": {
                    "type": "mrkdwn",
                    "text": f"*Retracted Release:* `{release_tag}`\n*Rolled back to:* `{previous_tag}`\n*Repository:* `{GITHUB_REPO}`"
                }
            },
            {
                "type": "section",
                "text": {
                    "type": "mrkdwn",
                    "text": f"*Action Required:* Do not deploy `{release_tag}`. Use `{previous_tag}` for all future deployments."
                }
            }
        ]
    }

    try:
        import requests
        response = requests.post(SLACK_WEBHOOK_URL, json=message, timeout=5)
        response.raise_for_status()
        log_audit("slack_notification_sent", {"release_tag": release_tag})
    except ImportError:
        print("requests library not installed, cannot send Slack notification")
    except Exception as e:
        print(f"Failed to send Slack notification: {str(e)}", file=sys.stderr)
        log_audit("slack_notification_failed", {"error": str(e)})

if __name__ == "__main__":
    # Validate required args
    if len(sys.argv) != 3:
        print(f"Usage: {sys.argv[0]}  ")
        sys.exit(1)

    retracted_tag = sys.argv[1]
    previous_tag = sys.argv[2]

    # Validate GITHUB_REPO format
    if not GITHUB_REPO or "/" not in GITHUB_REPO:
        print("Invalid GITHUB_REPO: must be in owner/repo format", file=sys.stderr)
        sys.exit(1)

    print(f"Starting retraction workflow for release {retracted_tag}, rolling back to {previous_tag}")

    try:
        # Step 1: Retract GitHub release
        retract_github_release(retracted_tag)

        # Step 2: Roll back Kubernetes deployment (assume revision matches previous tag)
        # In production, you'd map tags to revisions via a lookup table
        rollback_kubernetes_deployment(previous_tag)

        # Step 3: Notify stakeholders
        notify_stakeholders(retracted_tag, previous_tag)

        print(f"Retraction workflow completed successfully for {retracted_tag}")
        log_audit("retraction_workflow_completed", {
            "retracted_tag": retracted_tag,
            "previous_tag": previous_tag
        })
        sys.exit(0)
    except Exception as e:
        print(f"Retraction workflow failed: {str(e)}", file=sys.stderr)
        log_audit("retraction_workflow_failed", {"error": str(e)})
        sys.exit(1)

import os
import json
import datetime
import csv
from typing import List, Dict, Any, Optional
from dataclasses import dataclass

# Configuration
AUDIT_LOG_DIR = os.getenv("AUDIT_LOG_DIR", "/var/log/retraction")
GITHUB_REPO = os.getenv("GITHUB_REPO")
COMPLIANCE_REPORT_PATH = os.getenv("COMPLIANCE_REPORT_PATH", "./retraction_compliance_report.csv")
RETENTION_DAYS = int(os.getenv("RETENTION_DAYS", 365))

@dataclass
class RetractionEvent:
    """Structured representation of a retraction event."""
    event_id: str
    timestamp: datetime.datetime
    event_type: str
    release_tag: Optional[str]
    actor: str
    details: Dict[str, Any]
    audit_log_path: str

def load_audit_logs(audit_dir: str, days: int) -> List[Dict[str, Any]]:
    """Load audit log entries from the last N days."""
    cutoff = datetime.datetime.utcnow() - datetime.timedelta(days=days)
    log_entries = []

    if not os.path.isdir(audit_dir):
        raise ValueError(f"Audit log directory {audit_dir} does not exist")

    for filename in os.listdir(audit_dir):
        if not filename.endswith(".log"):
            continue
        filepath = os.path.join(audit_dir, filename)
        try:
            with open(filepath, "r") as f:
                for line_num, line in enumerate(f, 1):
                    line = line.strip()
                    if not line:
                        continue
                    try:
                        entry = json.loads(line)
                        # Parse timestamp
                        ts = datetime.datetime.fromisoformat(entry["timestamp"])
                        if ts >= cutoff:
                            log_entries.append({**entry, "audit_log_path": filepath, "line_num": line_num})
                    except json.JSONDecodeError as e:
                        print(f"Warning: Invalid JSON in {filepath} line {line_num}: {str(e)}")
                    except KeyError as e:
                        print(f"Warning: Missing key {str(e)} in {filepath} line {line_num}")
        except IOError as e:
            print(f"Warning: Failed to read audit log {filepath}: {str(e)}")

    return log_entries

def fetch_github_retraction_events(repo: str, days: int) -> List[Dict[str, Any]]:
    """Fetch retraction-related events from GitHub API (via gh CLI)."""
    if not repo:
        return []

    cutoff = datetime.datetime.utcnow() - datetime.timedelta(days=days)
    # Use gh CLI to fetch release events
    import subprocess
    cmd = [
        "gh", "api",
        f"/repos/{repo}/releases",
        "--jq", ".[] | select(.prerelease == true and (.body | contains("RETRACTED RELEASE"))) | {id, tag_name, published_at, body}"
    ]
    try:
        result = subprocess.run(cmd, capture_output=True, text=True, check=True)
        events = []
        for line in result.stdout.strip().split("\n"):
            if not line:
                continue
            release = json.loads(line)
            published_at = datetime.datetime.fromisoformat(release["published_at"].replace("Z", "+00:00"))
            if published_at >= cutoff:
                events.append({
                    "event_type": "github_release_retracted",
                    "release_tag": release["tag_name"],
                    "timestamp": published_at.isoformat(),
                    "details": {"release_id": release["id"], "body": release["body"]}
                })
        return events
    except subprocess.CalledProcessError as e:
        print(f"Warning: Failed to fetch GitHub events: {e.stderr}")
        return []
    except ImportError:
        print("Warning: subprocess not available, cannot fetch GitHub events")
        return []

def generate_compliance_report(events: List[RetractionEvent], output_path: str) -> None:
    """Generate a CSV compliance report for auditors."""
    fieldnames = [
        "event_id", "timestamp", "event_type", "release_tag", "actor",
        "deprecation_endpoint", "kubernetes_deployment", "client_id",
        "audit_log_path", "details_json"
    ]

    with open(output_path, "w", newline="") as csvfile:
        writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
        writer.writeheader()

        for event in events:
            # Flatten details for CSV
            details_json = json.dumps(event.details)
            row = {
                "event_id": event.event_id,
                "timestamp": event.timestamp.isoformat(),
                "event_type": event.event_type,
                "release_tag": event.release_tag or "",
                "actor": event.actor,
                "deprecation_endpoint": event.details.get("endpoint.path", ""),
                "kubernetes_deployment": event.details.get("deployment", ""),
                "client_id": event.details.get("client.id", ""),
                "audit_log_path": event.audit_log_path,
                "details_json": details_json
            }
            writer.writerow(row)

    print(f"Compliance report generated at {output_path}")

def main():
    # Validate configuration
    if not GITHUB_REPO:
        print("Warning: GITHUB_REPO not set, GitHub events will not be included")

    # Load audit logs
    print(f"Loading audit logs from {AUDIT_LOG_DIR} (last {RETENTION_DAYS} days)")
    audit_entries = load_audit_logs(AUDIT_LOG_DIR, RETENTION_DAYS)

    # Fetch GitHub events
    print(f"Fetching GitHub retraction events from {GITHUB_REPO}")
    github_events = fetch_github_retraction_events(GITHUB_REPO, RETENTION_DAYS)

    # Combine and deduplicate events
    all_events: List[RetractionEvent] = []
    seen_ids = set()

    # Process audit entries
    for entry in audit_entries:
        event_id = f"audit_{entry['timestamp']}_{entry['event_type']}"
        if event_id in seen_ids:
            continue
        seen_ids.add(event_id)
        all_events.append(RetractionEvent(
            event_id=event_id,
            timestamp=datetime.datetime.fromisoformat(entry["timestamp"]),
            event_type=entry["event_type"],
            release_tag=entry["details"].get("release_tag"),
            actor=entry["actor"],
            details=entry["details"],
            audit_log_path=entry["audit_log_path"]
        ))

    # Process GitHub events
    for event in github_events:
        event_id = f"github_{event['timestamp']}_{event['release_tag']}"
        if event_id in seen_ids:
            continue
        seen_ids.add(event_id)
        all_events.append(RetractionEvent(
            event_id=event_id,
            timestamp=datetime.datetime.fromisoformat(event["timestamp"]),
            event_type=event["event_type"],
            release_tag=event["release_tag"],
            actor="github",
            details=event["details"],
            audit_log_path="github_api"
        ))

    # Sort events by timestamp
    all_events.sort(key=lambda x: x.timestamp)

    # Generate report
    generate_compliance_report(all_events, COMPLIANCE_REPORT_PATH)

    # Print summary
    print(f"\nCompliance Report Summary:")
    print(f"Total retraction events: {len(all_events)}")
    print(f"Event types: {set(e.event_type for e in all_events)}")
    print(f"Date range: {all_events[0].timestamp.strftime('%Y-%m-%d')} to {all_events[-1].timestamp.strftime('%Y-%m-%d')}")

if __name__ == "__main__":
    main()

# OpenAPI 3.1.0 snippet for deprecated endpoint
paths:
  /v1/products:
    get:
      deprecated: true
      summary: Legacy product list (deprecated)
      responses:
        200:
          headers:
            X-API-Sunset-Date:
              schema:
                type: string
                format: date
                example: 2024-12-31
            X-API-Deprecation-Notice:
              schema:
                type: string
                example: "Sunset date: 2024-12-31. Use /v2/products instead."

// k6 v0.49.0 script to test deprecated endpoint headers
import http from 'k6/http';
import { check, sleep } from 'k6';

export const options = {
  vus: 100,
  duration: '30s',
};

export default function () {
  const res = http.get('https://staging-api.example.com/v1/products');
  check(res, {
    'status is 200': (r) => r.status === 200,
    'deprecation header exists': (r) => r.headers['X-API-Deprecation-Notice'] !== undefined,
    'sunset date is valid': (r) => Date.parse(r.headers['X-API-Sunset-Date']) > Date.now(),
    'link header points to replacement': (r) => r.headers['Link'].includes('/v2/products'),
  });
  sleep(1);
}

maker-retraction-workflows/
├── README.md                # Project overview and setup instructions
├── RUNBOOK.md               # On-call retraction runbook
├── requirements.txt         # Python dependencies for all scripts
├── middleware/
│   └── api_deprecation.py   # Code example 1: API deprecation middleware
├── retraction/
│   ├── release_retractor.py # Code example 2: Automated release retraction script
│   └── compliance_report.py # Code example 3: Compliance report generator
├── tests/
│   ├── test_middleware.py   # Unit tests for deprecation middleware
│   ├── test_retractor.py   # Unit tests for retraction script
│   └── k6/                 # k6 load test scripts
│       └── deprecated_endpoint_test.js
└── config/
    ├── deprecated_endpoints.json # Sample deprecated endpoint config
    └── otel_config.yaml         # OpenTelemetry configuration