spaceforge 0.0.3

A Python framework for building Spacelift plugins

Spaceforge - Build Spacelift Plugins in Python

Spaceforge is a Python framework that makes it easy to build powerful Spacelift plugins using a declarative, hook-based approach. Define your plugin logic in Python, and spaceforge automatically generates the plugin manifest for Spacelift.

Installation

Install spaceforge from PyPI:

pip install spaceforge

Quick Start

1. Create Your Plugin

Create a Python file (e.g., my_plugin.py) and inherit from SpaceforgePlugin:

from spaceforge import SpaceforgePlugin, Parameter, Variable, Context
import os

class MyPlugin(SpaceforgePlugin):
    # Plugin metadata
    __plugin_name__ = "my-awesome-plugin"
    __version__ = "1.0.0"
    __author__ = "Your Name"
    
    # Define plugin parameters
    __parameters__ = [
        Parameter(
            name="api_key",
            description="API key for external service",
            required=True,
            sensitive=True
        ),
        Parameter(
            name="environment",
            description="Target environment",
            required=False,
            default="production"
        )
    ]
    
    # Define Spacelift contexts
    __contexts__ = [
        Context(
            name_prefix="my-plugin",
            description="Main plugin context",
            env=[
                Variable(
                    key="API_KEY",
                    value_from_parameter="api_key",
                    sensitive=True
                ),
                Variable(
                    key="ENVIRONMENT",
                    value_from_parameter="environment"
                )
            ]
        )
    ]
    
    def after_plan(self):
        """Run security checks after Terraform plan"""
        # Run external commands
        return_code, stdout, stderr = self.run_cli("my-security-tool", "--scan", "./", '--api', os.environ["API_KEY"])
        
        if return_code != 0:
            self.logger.error("Security scan failed!")
            exit(1)
            
        self.logger.info("Security scan passed!")

2. Generate Plugin Manifest

Generate the Spacelift plugin YAML manifest:

spaceforge generate my_plugin.py

This creates plugin.yaml that you can upload to Spacelift.

3. Test Your Plugin

Test individual hooks locally:

# Set parameter values
export API_KEY="your-api-key"
export ENVIRONMENT="staging"

# Test the after_plan hook
spaceforge runner after_plan

Available Hooks

Override these methods in your plugin to add custom logic:

• before_init() - Before Terraform init
• after_init() - After Terraform init
• before_plan() - Before Terraform plan
• after_plan() - After Terraform plan
• before_apply() - Before Terraform apply
• after_apply() - After Terraform apply
• before_perform() - Before the run performs
• after_perform() - After the run performs
• before_destroy() - Before Terraform destroy
• after_destroy() - After Terraform destroy
• after_run() - After the run completes

Plugin Components

Parameters

Define user-configurable parameters:

__parameters__ = [
    Parameter(
        name="database_url",
        description="Database connection URL",
        required=True,
        sensitive=True,
        default="postgresql://localhost:5432/mydb"
    ),
    Parameter(
        name="timeout", 
        description="Timeout in seconds",
        required=False,
        default=30
    )
]

Contexts

Define Spacelift contexts with environment variables and custom hooks:

__contexts__ = [
    Context(
        name_prefix="production",
        description="Production environment context",
        labels={"env": "prod"},
        env=[
            Variable(
                key="DATABASE_URL",
                value_from_parameter="database_url",
                sensitive=True
            ),
            Variable(
                key="API_ENDPOINT", 
                value="https://api.prod.example.com"
            )
        ],
        hooks={
            "before_apply": [
                "echo 'Starting production deployment'",
                "kubectl get pods"
            ]
        }
    )
]

Binaries

Automatically download and install external tools:

__binaries__ = [
    Binary(
        name="kubectl",
        download_urls={
            "amd64": "https://dl.k8s.io/release/v1.28.0/bin/linux/amd64/kubectl",
            "arm64": "https://dl.k8s.io/release/v1.28.0/bin/linux/arm64/kubectl"
        }
    )
]

Policies

Define OPA policies for your plugin:

__policies__ = [
    Policy(
        name_prefix="security-check",
        type="notification",
        body="""
package spacelift

webhook[{"endpoint_id": "security-alerts"}] {
  input.run_updated.run.marked_unsafe == true
}
        """,
        labels={"type": "security"}
    )
]

Webhooks

Define webhooks to trigger external actions:

__webhooks__ = [
    Webhook(
        name_prefix="security-alerts",
        description="Send security alerts to external service",
        endpoint="https://alerts.example.com/webhook",
        secrets=[
            Variable(key="amazing", value="secret-value", sensitive=True)
        ],
    )
]

Plugin Features

Logging

Built-in structured logging with run context:

def after_plan(self):
    self.logger.info("Starting security scan")
    self.logger.debug("Debug info (only shown when SPACELIFT_DEBUG=true)")  
    self.logger.warning("Warning message")
    self.logger.error("Error occurred")

CLI Execution

Run external commands with automatic logging:

def before_apply(self):
    # Run command with automatic output capture
    return_code, stdout, stderr = self.run_cli("terraform", "validate")
    
    if return_code != 0:
        self.logger.error("Terraform validation failed")
        exit(1)

Spacelift API Integration

Query the Spacelift GraphQL API (requires SPACELIFT_API_TOKEN and SPACELIFT_DOMAIN):

def after_plan(self):
    result = self.query_api("""
        query {
            stack(id: "my-stack-id") {
                name
                state
                latestRun {
                    id
                    state
                }
            }
        }
    """)
    
    self.logger.info(f"Stack state: {result['stack']['state']}")

Access Plan and State

Access Terraform plan and state data:

def after_plan(self):
    # Get the current plan
    plan = self.get_plan_json()
    
    # Get the state before changes
    state = self.get_state_before_json()
    
    # Analyze planned changes
    resource_count = len(plan.get('planned_values', {}).get('root_module', {}).get('resources', []))
    self.logger.info(f"Planning to manage {resource_count} resources")

Send Rich Output

Send formatted markdown to the Spacelift UI:

def after_plan(self):
    markdown = """
    # Security Scan Results
    
    ✅ **Passed:** 45 checks
    ⚠️ **Warnings:** 3 issues  
    ❌ **Failed:** 0 critical issues
    
    [View detailed report](https://security.example.com/reports/123)
    """
    
    self.send_markdown(markdown)

Append to Policy Input

Append custom data to the OPA policy input:

The following example will create input available via input.third_party_metadata.custom.my_custom_data in your OPA policies:

def after_plan(self):
    self.append_policy_input("my_custom_data", {
        "scan_results": {
            "passed": True,
            "issues": []
        }
    })

CLI Commands

Generate Plugin Manifest

# Generate from plugin.py (default filename)
spaceforge generate

# Generate from specific file  
spaceforge generate my_plugin.py

# Specify output file
spaceforge generate my_plugin.py -o custom-output.yaml

# Get help
spaceforge generate --help

Test Plugin Hooks

# Set parameters via environment variables
export SPACEFORGE_PARAM_API_KEY="test-key" 
export SPACEFORGE_PARAM_TIMEOUT="60"

# Test specific hook
spaceforge runner after_plan

# Test with specific plugin file
spaceforge runner --plugin-file my_plugin.py before_apply

# Get help
spaceforge runner --help

Plugin Development Tips

1. Handle Dependencies

If your plugin needs Python packages, create a requirements.txt file. Spaceforge automatically adds a before_init hook to install them:

requests>=2.28.0
pydantic>=1.10.0

2. Environment Variables

Access Spacelift environment variables in your hooks:

def after_plan(self):
    run_id = os.environ.get('SPACELIFT_RUN_ID')
    stack_id = os.environ.get('SPACELIFT_STACK_ID')
    self.logger.info(f"Processing run {run_id} for stack {stack_id}")

3. Error Handling

Always handle errors gracefully:

def after_plan(self):
    try:
        # Your plugin logic here
        result = self.run_external_service()
        
    except Exception as e:
        self.logger.error(f"Plugin failed: {str(e)}")
        # Exit with non-zero code to fail the run
        exit(1)

4. Testing and Debugging

• Set SPACELIFT_DEBUG=true to enable debug logging
• Use the runner command to test hooks during development
• Test with different parameter combinations
• Validate your generated YAML before uploading to Spacelift

Example: Security Scanning Plugin

Here's a complete example of a security scanning plugin:

import os
import json
from spaceforge import SpaceforgePlugin, Parameter, Variable, Context, Binary

class SecurityScannerPlugin(SpaceforgePlugin):
    __plugin_name__ = "security-scanner"
    __version__ = "1.0.0"
    __author__ = "Security Team"
    
    __binaries__ = [
        Binary(
            name="security-cli",
            download_urls={
                "amd64": "https://releases.example.com/security-cli-linux-amd64",
                "arm64": "https://releases.example.com/security-cli-linux-arm64"
            }
        )
    ]
    
    __parameters__ = [
        Parameter(
            name="api_token",
            description="Security service API token",
            required=True,
            sensitive=True
        ),
        Parameter(
            name="severity_threshold", 
            description="Minimum severity level to report",
            required=False,
            default="medium"
        )
    ]
    
    __contexts__ = [
        Context(
            name_prefix="security-scanner",
            description="Security scanning context",
            env=[
                Variable(
                    key="SECURITY_API_TOKEN",
                    value_from_parameter="api_token",
                    sensitive=True
                ),
                Variable(
                    key="SEVERITY_THRESHOLD",
                    value_from_parameter="severity_threshold"
                )
            ]
        )
    ]
    
    def after_plan(self):
        """Run security scan after Terraform plan"""
        self.logger.info("Starting security scan of Terraform plan")
        
        # Authenticate with security service
        return_code, stdout, stderr = self.run_cli(
            "security-cli", "auth", 
            "--token", os.environ["SECURITY_API_TOKEN"]
        )
        
        if return_code != 0:
            self.logger.error("Failed to authenticate with security service")
            exit(1)
        
        # Scan the Terraform plan
        return_code, stdout, stderr = self.run_cli(
            "security-cli", "scan", "terraform", 
            "--plan-file", "spacelift.plan.json",
            "--format", "json",
            "--severity", os.environ.get("SEVERITY_THRESHOLD", "medium"),
            print_output=False
        )
        
        if return_code != 0:
            self.logger.error("Security scan failed")
            for line in stderr:
                self.logger.error(line)
            exit(1)
        
        # Parse scan results
        try:
            results = json.loads('\n'.join(stdout))
            
            # Generate markdown report
            markdown = self._generate_report(results)
            self.send_markdown(markdown)
            
            # Fail run if critical issues found
            if results.get('critical_count', 0) > 0:
                self.logger.error(f"Found {results['critical_count']} critical security issues")
                exit(1)
                
            self.logger.info("Security scan completed successfully")
            
        except json.JSONDecodeError:
            self.logger.error("Failed to parse scan results")
            exit(1)
    
    def _generate_report(self, results):
        """Generate markdown report from scan results"""
        report = "# Security Scan Results\n\n"
        
        if results.get('total_issues', 0) == 0:
            report += "✅ **No security issues found!**\n"
        else:
            report += f"Found {results['total_issues']} security issues:\n\n"
            
            for severity in ['critical', 'high', 'medium', 'low']:
                count = results.get(f'{severity}_count', 0)
                if count > 0:
                    emoji = {'critical': '🔴', 'high': '🟠', 'medium': '🟡', 'low': '🟢'}[severity]
                    report += f"- {emoji} **{severity.upper()}:** {count}\n"
        
        if results.get('report_url'):
            report += f"\n[View detailed report]({results['report_url']})\n"
            
        return report

Generate and test this plugin:

# Generate the manifest
spaceforge generate security_scanner.py

# Test locally
export SPACEFORGE_PARAM_API_TOKEN="your-token"
export SPACEFORGE_PARAM_SEVERITY_THRESHOLD="high"
spaceforge runner after_plan

Next Steps

1. Install spaceforge: pip install spaceforge
2. Create your plugin: Start with the quick start example
3. Test locally: Use the runner command to test your hooks
4. Generate manifest: Use the generate command to create plugin.yaml
5. Upload to Spacelift: Add your plugin manifest to your Spacelift account

For more advanced examples, see the plugins directory in this repository.