# 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:
```bash
pip install spaceforge
```
## Quick Start
### 1. Create Your Plugin
Create a Python file (e.g., `my_plugin.py`) and inherit from `SpaceforgePlugin`:
```python
from spaceforge import SpaceforgePlugin, Parameter, Variable, Context, Binary, Policy, Webhook, MountedFile
import os
class MyPlugin(SpaceforgePlugin):
# Plugin metadata
__plugin_name__ = "my-awesome-plugin"
__version__ = "1.0.0"
__author__ = "Your Name"
__labels__ = ["security", "monitoring"] # Optional labels for categorization
# Define plugin parameters
__parameters__ = [
Parameter(
name="API Key",
id="api_key", # Optional ID for parameter reference
description="API key for external service",
required=True,
sensitive=True
),
Parameter(
name="Environment",
id="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", # Matches parameter id or name
sensitive=True
),
Variable(
key="ENVIRONMENT",
value_from_parameter="environment" # Matches parameter id or name
)
]
)
]
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:
```bash
spaceforge generate my_plugin.py
```
This creates `plugin.yaml` that you can upload to Spacelift.
### 3. Test Your Plugin
Test individual hooks locally:
```bash
# 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
### Labels
Add optional labels to categorize your plugin:
```python
class MyPlugin(SpaceforgePlugin):
__labels__ = ["security", "monitoring", "compliance"]
```
### Parameters
Define user-configurable parameters:
```python
__parameters__ = [
Parameter(
name="Database URL",
id="database_url", # Optional: used for parameter reference
description="Database connection URL",
required=True,
sensitive=True
),
Parameter(
name="Timeout",
id="timeout",
description="Timeout in seconds",
required=False,
default="30" # Default values should be strings
)
]
```
**Parameter Notes:**
- Parameter `name` is displayed in the Spacelift UI
- Parameter `id` (optional) is used for programmatic reference
- `value_from_parameter` can reference either the `id` (if present) or the `name`
- Parameters are made available as environment variables through Variable definitions
- Default values must be strings
- Required parameters cannot have default values
### Contexts
Define Spacelift contexts with environment variables and custom hooks:
```python
__contexts__ = [
Context(
name_prefix="production",
description="Production environment context",
labels=["env:prod"],
env=[
Variable(
key="DATABASE_URL",
value_from_parameter="database_url", # Matches parameter id
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:
```python
__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"
}
)
]
```
**Context Priority System:**
Control the execution order of contexts using the `priority` field:
```python
__contexts__ = [
Context(
name_prefix="setup",
description="Setup context (runs first)",
priority=0, # Lower numbers run first
hooks={
"before_init": ["echo 'Setting up environment'"]
}
),
Context(
name_prefix="main",
description="Main context (runs second)",
priority=1, # Higher numbers run after lower ones
hooks={
"before_init": ["echo 'Main execution'"]
}
)
]
```
**Priority Notes:**
- Default priority is `0`
- Lower numbers execute first (0, then 1, then 2, etc.)
- Useful for ensuring setup contexts run before main execution contexts
**Binary PATH Management:**
- When using Python hook methods (e.g., `def before_apply()`), binaries are automatically available in PATH
- When using raw context hooks, you must manually export the PATH:
```python
__contexts__ = [
Context(
name_prefix="kubectl-setup",
description="Setup kubectl binary for raw hooks",
hooks={
"before_init": [
'export PATH="/mnt/workspace/plugins/plugin_binaries:$PATH"',
"kubectl version"
]
}
)
]
```
### Mounted Files
Mount file content directly into contexts:
```python
from spaceforge import MountedFile
__contexts__ = [
Context(
name_prefix="config",
description="Context with mounted configuration files",
mounted_files=[
MountedFile(
path="tmp/config.json",
content='{"environment": "production", "debug": false}',
sensitive=False
),
MountedFile(
path="tmp/secret-config.yaml",
content="api_key: secret-value\nendpoint: https://api.example.com",
sensitive=True # Marks content as sensitive
)
]
)
]
```
**MountedFile Notes:**
- Files are created at the specified path when the context is applied
- Content is written exactly as provided
- Use `sensitive=True` for files containing secrets or sensitive data
- path is from `/mnt/workspace/`. An example would be `tmp/config.json` which would be mounted at `/mnt/workspace/tmp/config.json`
### Policies
Define OPA policies for your plugin:
```python
__policies__ = [
Policy(
name_prefix="security-check",
type="NOTIFICATION",
body="""
package spacelift
webhook[{"endpoint_id": "security-alerts"}] {
input.run_updated.run.marked_unsafe == true
}
""",
labels=["security"]
)
]
```
### Webhooks
Define webhooks to trigger external actions:
```python
__webhooks__ = [
Webhook(
name_prefix="security-alerts",
endpoint="https://alerts.example.com/webhook",
secretFromParameter="webhook_secret", # Parameter id/name for webhook secret
labels=["security"]
)
]
```
## Plugin Features
### Logging
Built-in structured logging with run context:
```python
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:
```python
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 `TF_VAR_spacelift_graphql_endpoint`):
```python
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']}")
```
### User Token Authentication
Use user API tokens instead of service tokens for Spacelift API access. This is useful because the token on the run may not have sufficient permissions for certain operations.
```python
def before_plan(self):
# Use user API token for authentication
user_id = os.environ.get('SPACELIFT_USER_ID')
user_secret = os.environ.get('SPACELIFT_USER_SECRET')
if user_id and user_secret:
self.use_user_token(user_id, user_secret)
# Now you can use the API with user permissions
result = self.query_api("""
query {
viewer {
id
login
}
}
""")
self.logger.info(f"Authenticated as: {result['viewer']['login']}")
```
**User Token Notes:**
- Allows plugins to act on behalf of a specific user
- Useful for operations requiring user-specific permissions
- User tokens may have different access levels than service tokens
- Call `use_user_token()` before making API requests
### Access Plan and State
Access Terraform plan and state data:
```python
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:
```python
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)
```
### Add to Policy Input
Add 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:
```python
def after_plan(self):
self.add_to_policy_input("my_custom_data", {
"scan_results": {
"passed": True,
"issues": []
}
})
```
## CLI Commands
### Generate Plugin Manifest
```bash
# 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
```bash
# Set parameters for local testing (parameters are normally provided by Spacelift)
export API_KEY="test-key"
export 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:
```txt
requests>=2.28.0
pydantic>=1.10.0
```
### 2. Environment Variables
Access Spacelift environment variables in your hooks:
```python
def after_plan(self):
run_id = os.environ.get('TF_VAR_spacelift_run_id')
stack_id = os.environ.get('TF_VAR_spacelift_stack_id')
self.logger.info(f"Processing run {run_id} for stack {stack_id}")
```
### 3. Error Handling
Always handle errors gracefully:
```python
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:
```python
import os
import json
from spaceforge import SpaceforgePlugin, Parameter, Variable, Context, Binary, Policy, MountedFile
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",
id="api_token",
description="Security service API token",
required=True,
sensitive=True
),
Parameter(
name="Severity Threshold",
id="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:
```bash
# Generate the manifest
spaceforge generate security_scanner.py
# Test locally
export API_TOKEN="your-token"
export SEVERITY_THRESHOLD="high"
spaceforge runner after_plan
```
## Speeding up plugin execution
There are a few things you can do to speed up plugin execution.
1. Ensure your runner has `spaceforge` preinstalled. This will avoid the overhead of installing it during the run. (15-30 seconds)
2. If youre using binaries, we will only install the binary if its not found. You can gain a few seconds by ensuring its already on the runner.
3. If your plugin has a lot of dependencies, consider using a prebuilt runner image with your plugin and its dependencies installed. This avoids the overhead of installing them during each run.
4. Ensure your runner has enough core resources (CPU, memory) to handle the plugin execution efficiently. If your plugin is resource-intensive, consider using a more powerful runner.
## 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](plugins/) directory in this repository.
Raw data
{
"_id": null,
"home_page": "https://github.com/spacelift-io/plugins",
"name": "spaceforge",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.9",
"maintainer_email": "Spacelift <support@spacelift.io>",
"keywords": "spacelift, plugin, framework, infrastructure, devops, spaceforge",
"author": "Spacelift",
"author_email": "Spacelift <support@spacelift.io>",
"download_url": "https://files.pythonhosted.org/packages/b9/a7/767245871be47d73c2a17004cbc6be6753ef7cd99ba4def0057ee752dd95/spaceforge-1.1.2.tar.gz",
"platform": null,
"description": "# Spaceforge - Build Spacelift Plugins in Python\n\nSpaceforge 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.\n\n## Installation\n\nInstall spaceforge from PyPI:\n\n```bash\npip install spaceforge\n```\n\n## Quick Start\n\n### 1. Create Your Plugin\n\nCreate a Python file (e.g., `my_plugin.py`) and inherit from `SpaceforgePlugin`:\n\n```python\nfrom spaceforge import SpaceforgePlugin, Parameter, Variable, Context, Binary, Policy, Webhook, MountedFile\nimport os\n\nclass MyPlugin(SpaceforgePlugin):\n # Plugin metadata\n __plugin_name__ = \"my-awesome-plugin\"\n __version__ = \"1.0.0\"\n __author__ = \"Your Name\"\n __labels__ = [\"security\", \"monitoring\"] # Optional labels for categorization\n \n # Define plugin parameters\n __parameters__ = [\n Parameter(\n name=\"API Key\",\n id=\"api_key\", # Optional ID for parameter reference\n description=\"API key for external service\",\n required=True,\n sensitive=True\n ),\n Parameter(\n name=\"Environment\",\n id=\"environment\",\n description=\"Target environment\",\n required=False,\n default=\"production\"\n )\n ]\n \n # Define Spacelift contexts\n __contexts__ = [\n Context(\n name_prefix=\"my-plugin\",\n description=\"Main plugin context\",\n env=[\n Variable(\n key=\"API_KEY\",\n value_from_parameter=\"api_key\", # Matches parameter id or name\n sensitive=True\n ),\n Variable(\n key=\"ENVIRONMENT\",\n value_from_parameter=\"environment\" # Matches parameter id or name\n )\n ]\n )\n ]\n \n def after_plan(self):\n \"\"\"Run security checks after Terraform plan\"\"\"\n # Run external commands\n return_code, stdout, stderr = self.run_cli(\"my-security-tool\", \"--scan\", \"./\", '--api', os.environ[\"API_KEY\"])\n \n if return_code != 0:\n self.logger.error(\"Security scan failed!\")\n exit(1)\n \n self.logger.info(\"Security scan passed!\")\n```\n\n### 2. Generate Plugin Manifest\n\nGenerate the Spacelift plugin YAML manifest:\n\n```bash\nspaceforge generate my_plugin.py\n```\n\nThis creates `plugin.yaml` that you can upload to Spacelift.\n\n### 3. Test Your Plugin\n\nTest individual hooks locally:\n\n```bash\n# Set parameter values\nexport API_KEY=\"your-api-key\"\nexport ENVIRONMENT=\"staging\"\n\n# Test the after_plan hook\nspaceforge runner after_plan\n```\n\n## Available Hooks\n\nOverride these methods in your plugin to add custom logic:\n\n- `before_init()` - Before Terraform init\n- `after_init()` - After Terraform init \n- `before_plan()` - Before Terraform plan\n- `after_plan()` - After Terraform plan\n- `before_apply()` - Before Terraform apply\n- `after_apply()` - After Terraform apply\n- `before_perform()` - Before the run performs\n- `after_perform()` - After the run performs\n- `before_destroy()` - Before Terraform destroy\n- `after_destroy()` - After Terraform destroy\n- `after_run()` - After the run completes\n\n## Plugin Components\n\n### Labels\n\nAdd optional labels to categorize your plugin:\n\n```python\nclass MyPlugin(SpaceforgePlugin):\n __labels__ = [\"security\", \"monitoring\", \"compliance\"]\n```\n\n### Parameters\n\nDefine user-configurable parameters:\n\n```python\n__parameters__ = [\n Parameter(\n name=\"Database URL\",\n id=\"database_url\", # Optional: used for parameter reference\n description=\"Database connection URL\",\n required=True,\n sensitive=True\n ),\n Parameter(\n name=\"Timeout\", \n id=\"timeout\",\n description=\"Timeout in seconds\",\n required=False,\n default=\"30\" # Default values should be strings\n )\n]\n```\n\n**Parameter Notes:**\n- Parameter `name` is displayed in the Spacelift UI\n- Parameter `id` (optional) is used for programmatic reference\n- `value_from_parameter` can reference either the `id` (if present) or the `name`\n- Parameters are made available as environment variables through Variable definitions\n- Default values must be strings\n- Required parameters cannot have default values\n\n### Contexts\n\nDefine Spacelift contexts with environment variables and custom hooks:\n\n```python\n__contexts__ = [\n Context(\n name_prefix=\"production\",\n description=\"Production environment context\",\n labels=[\"env:prod\"],\n env=[\n Variable(\n key=\"DATABASE_URL\",\n value_from_parameter=\"database_url\", # Matches parameter id\n sensitive=True\n ),\n Variable(\n key=\"API_ENDPOINT\", \n value=\"https://api.prod.example.com\"\n )\n ],\n hooks={\n \"before_apply\": [\n \"echo 'Starting production deployment'\",\n \"kubectl get pods\"\n ]\n }\n )\n]\n```\n\n### Binaries\n\nAutomatically download and install external tools:\n\n```python\n__binaries__ = [\n Binary(\n name=\"kubectl\",\n download_urls={\n \"amd64\": \"https://dl.k8s.io/release/v1.28.0/bin/linux/amd64/kubectl\",\n \"arm64\": \"https://dl.k8s.io/release/v1.28.0/bin/linux/arm64/kubectl\"\n }\n )\n]\n```\n\n**Context Priority System:**\n\nControl the execution order of contexts using the `priority` field:\n\n```python\n__contexts__ = [\n Context(\n name_prefix=\"setup\",\n description=\"Setup context (runs first)\",\n priority=0, # Lower numbers run first\n hooks={\n \"before_init\": [\"echo 'Setting up environment'\"]\n }\n ),\n Context(\n name_prefix=\"main\", \n description=\"Main context (runs second)\",\n priority=1, # Higher numbers run after lower ones\n hooks={\n \"before_init\": [\"echo 'Main execution'\"]\n }\n )\n]\n```\n\n**Priority Notes:**\n- Default priority is `0`\n- Lower numbers execute first (0, then 1, then 2, etc.)\n- Useful for ensuring setup contexts run before main execution contexts\n\n**Binary PATH Management:**\n- When using Python hook methods (e.g., `def before_apply()`), binaries are automatically available in PATH\n- When using raw context hooks, you must manually export the PATH:\n\n```python\n__contexts__ = [\n Context(\n name_prefix=\"kubectl-setup\",\n description=\"Setup kubectl binary for raw hooks\",\n hooks={\n \"before_init\": [\n 'export PATH=\"/mnt/workspace/plugins/plugin_binaries:$PATH\"',\n \"kubectl version\"\n ]\n }\n )\n]\n```\n\n### Mounted Files\n\nMount file content directly into contexts:\n\n```python\nfrom spaceforge import MountedFile\n\n__contexts__ = [\n Context(\n name_prefix=\"config\",\n description=\"Context with mounted configuration files\",\n mounted_files=[\n MountedFile(\n path=\"tmp/config.json\",\n content='{\"environment\": \"production\", \"debug\": false}',\n sensitive=False\n ),\n MountedFile(\n path=\"tmp/secret-config.yaml\",\n content=\"api_key: secret-value\\nendpoint: https://api.example.com\",\n sensitive=True # Marks content as sensitive\n )\n ]\n )\n]\n```\n\n**MountedFile Notes:**\n- Files are created at the specified path when the context is applied\n- Content is written exactly as provided\n- Use `sensitive=True` for files containing secrets or sensitive data\n- path is from `/mnt/workspace/`. An example would be `tmp/config.json` which would be mounted at `/mnt/workspace/tmp/config.json`\n\n### Policies\n\nDefine OPA policies for your plugin:\n\n```python\n__policies__ = [\n Policy(\n name_prefix=\"security-check\",\n type=\"NOTIFICATION\",\n body=\"\"\"\npackage spacelift\n\nwebhook[{\"endpoint_id\": \"security-alerts\"}] {\n input.run_updated.run.marked_unsafe == true\n}\n \"\"\",\n labels=[\"security\"]\n )\n]\n```\n\n### Webhooks\n\nDefine webhooks to trigger external actions:\n\n```python\n__webhooks__ = [\n Webhook(\n name_prefix=\"security-alerts\",\n endpoint=\"https://alerts.example.com/webhook\",\n secretFromParameter=\"webhook_secret\", # Parameter id/name for webhook secret\n labels=[\"security\"]\n )\n]\n```\n\n## Plugin Features\n\n### Logging\n\nBuilt-in structured logging with run context:\n\n```python\ndef after_plan(self):\n self.logger.info(\"Starting security scan\")\n self.logger.debug(\"Debug info (only shown when SPACELIFT_DEBUG=true)\") \n self.logger.warning(\"Warning message\")\n self.logger.error(\"Error occurred\")\n```\n\n### CLI Execution\n\nRun external commands with automatic logging:\n\n```python\ndef before_apply(self):\n # Run command with automatic output capture\n return_code, stdout, stderr = self.run_cli(\"terraform\", \"validate\")\n \n if return_code != 0:\n self.logger.error(\"Terraform validation failed\")\n exit(1)\n```\n\n### Spacelift API Integration\n\nQuery the Spacelift GraphQL API (requires `SPACELIFT_API_TOKEN` and `TF_VAR_spacelift_graphql_endpoint`):\n\n```python\ndef after_plan(self):\n result = self.query_api(\"\"\"\n query {\n stack(id: \"my-stack-id\") {\n name\n state\n latestRun {\n id\n state\n }\n }\n }\n \"\"\")\n \n self.logger.info(f\"Stack state: {result['stack']['state']}\")\n```\n\n### User Token Authentication\n\nUse user API tokens instead of service tokens for Spacelift API access. This is useful because the token on the run may not have sufficient permissions for certain operations.\n\n```python\ndef before_plan(self):\n # Use user API token for authentication\n user_id = os.environ.get('SPACELIFT_USER_ID')\n user_secret = os.environ.get('SPACELIFT_USER_SECRET')\n \n if user_id and user_secret:\n self.use_user_token(user_id, user_secret)\n \n # Now you can use the API with user permissions\n result = self.query_api(\"\"\"\n query {\n viewer {\n id\n login\n }\n }\n \"\"\")\n \n self.logger.info(f\"Authenticated as: {result['viewer']['login']}\")\n```\n\n**User Token Notes:**\n- Allows plugins to act on behalf of a specific user\n- Useful for operations requiring user-specific permissions\n- User tokens may have different access levels than service tokens\n- Call `use_user_token()` before making API requests\n\n### Access Plan and State\n\nAccess Terraform plan and state data:\n\n```python\ndef after_plan(self):\n # Get the current plan\n plan = self.get_plan_json()\n \n # Get the state before changes\n state = self.get_state_before_json()\n \n # Analyze planned changes\n resource_count = len(plan.get('planned_values', {}).get('root_module', {}).get('resources', []))\n self.logger.info(f\"Planning to manage {resource_count} resources\")\n```\n\n### Send Rich Output\n\nSend formatted markdown to the Spacelift UI:\n\n```python\ndef after_plan(self):\n markdown = \"\"\"\n # Security Scan Results\n \n \u2705 **Passed:** 45 checks\n \u26a0\ufe0f **Warnings:** 3 issues \n \u274c **Failed:** 0 critical issues\n \n [View detailed report](https://security.example.com/reports/123)\n \"\"\"\n \n self.send_markdown(markdown)\n```\n\n### Add to Policy Input\n\nAdd custom data to the OPA policy input:\n\nThe following example will create input available via `input.third_party_metadata.custom.my_custom_data` in your OPA policies:\n```python\ndef after_plan(self):\n self.add_to_policy_input(\"my_custom_data\", {\n \"scan_results\": {\n \"passed\": True,\n \"issues\": []\n }\n })\n```\n\n## CLI Commands\n\n### Generate Plugin Manifest\n\n```bash\n# Generate from plugin.py (default filename)\nspaceforge generate\n\n# Generate from specific file \nspaceforge generate my_plugin.py\n\n# Specify output file\nspaceforge generate my_plugin.py -o custom-output.yaml\n\n# Get help\nspaceforge generate --help\n```\n\n### Test Plugin Hooks\n\n```bash\n# Set parameters for local testing (parameters are normally provided by Spacelift)\nexport API_KEY=\"test-key\" \nexport TIMEOUT=\"60\"\n\n# Test specific hook\nspaceforge runner after_plan\n\n# Test with specific plugin file\nspaceforge runner --plugin-file my_plugin.py before_apply\n\n# Get help\nspaceforge runner --help\n```\n\n## Plugin Development Tips\n\n### 1. Handle Dependencies\n\nIf your plugin needs Python packages, create a `requirements.txt` file. Spaceforge automatically adds a `before_init` hook to install them:\n\n```txt\nrequests>=2.28.0\npydantic>=1.10.0\n```\n\n### 2. Environment Variables\n\nAccess Spacelift environment variables in your hooks:\n\n```python\ndef after_plan(self):\n run_id = os.environ.get('TF_VAR_spacelift_run_id')\n stack_id = os.environ.get('TF_VAR_spacelift_stack_id') \n self.logger.info(f\"Processing run {run_id} for stack {stack_id}\")\n```\n\n### 3. Error Handling\n\nAlways handle errors gracefully:\n\n```python\ndef after_plan(self):\n try:\n # Your plugin logic here\n result = self.run_external_service()\n \n except Exception as e:\n self.logger.error(f\"Plugin failed: {str(e)}\")\n # Exit with non-zero code to fail the run\n exit(1)\n```\n\n### 4. Testing and Debugging\n\n- Set `SPACELIFT_DEBUG=true` to enable debug logging\n- Use the runner command to test hooks during development\n- Test with different parameter combinations\n- Validate your generated YAML before uploading to Spacelift\n\n## Example: Security Scanning Plugin\n\nHere's a complete example of a security scanning plugin:\n\n```python\nimport os\nimport json\nfrom spaceforge import SpaceforgePlugin, Parameter, Variable, Context, Binary, Policy, MountedFile\n\nclass SecurityScannerPlugin(SpaceforgePlugin):\n __plugin_name__ = \"security-scanner\"\n __version__ = \"1.0.0\"\n __author__ = \"Security Team\"\n \n __binaries__ = [\n Binary(\n name=\"security-cli\",\n download_urls={\n \"amd64\": \"https://releases.example.com/security-cli-linux-amd64\",\n \"arm64\": \"https://releases.example.com/security-cli-linux-arm64\"\n }\n )\n ]\n \n __parameters__ = [\n Parameter(\n name=\"API Token\",\n id=\"api_token\",\n description=\"Security service API token\",\n required=True,\n sensitive=True\n ),\n Parameter(\n name=\"Severity Threshold\", \n id=\"severity_threshold\",\n description=\"Minimum severity level to report\",\n required=False,\n default=\"medium\"\n )\n ]\n \n __contexts__ = [\n Context(\n name_prefix=\"security-scanner\",\n description=\"Security scanning context\",\n env=[\n Variable(\n key=\"SECURITY_API_TOKEN\",\n value_from_parameter=\"api_token\",\n sensitive=True\n ),\n Variable(\n key=\"SEVERITY_THRESHOLD\",\n value_from_parameter=\"severity_threshold\"\n )\n ]\n )\n ]\n \n def after_plan(self):\n \"\"\"Run security scan after Terraform plan\"\"\"\n self.logger.info(\"Starting security scan of Terraform plan\")\n \n # Authenticate with security service\n return_code, stdout, stderr = self.run_cli(\n \"security-cli\", \"auth\", \n \"--token\", os.environ[\"SECURITY_API_TOKEN\"]\n )\n \n if return_code != 0:\n self.logger.error(\"Failed to authenticate with security service\")\n exit(1)\n \n # Scan the Terraform plan\n return_code, stdout, stderr = self.run_cli(\n \"security-cli\", \"scan\", \"terraform\", \n \"--plan-file\", \"spacelift.plan.json\",\n \"--format\", \"json\",\n \"--severity\", os.environ.get(\"SEVERITY_THRESHOLD\", \"medium\"),\n print_output=False\n )\n \n if return_code != 0:\n self.logger.error(\"Security scan failed\")\n for line in stderr:\n self.logger.error(line)\n exit(1)\n \n # Parse scan results\n try:\n results = json.loads('\\n'.join(stdout))\n \n # Generate markdown report\n markdown = self._generate_report(results)\n self.send_markdown(markdown)\n \n # Fail run if critical issues found\n if results.get('critical_count', 0) > 0:\n self.logger.error(f\"Found {results['critical_count']} critical security issues\")\n exit(1)\n \n self.logger.info(\"Security scan completed successfully\")\n \n except json.JSONDecodeError:\n self.logger.error(\"Failed to parse scan results\")\n exit(1)\n \n def _generate_report(self, results):\n \"\"\"Generate markdown report from scan results\"\"\"\n report = \"# Security Scan Results\\n\\n\"\n \n if results.get('total_issues', 0) == 0:\n report += \"\u2705 **No security issues found!**\\n\"\n else:\n report += f\"Found {results['total_issues']} security issues:\\n\\n\"\n \n for severity in ['critical', 'high', 'medium', 'low']:\n count = results.get(f'{severity}_count', 0)\n if count > 0:\n emoji = {'critical': '\ud83d\udd34', 'high': '\ud83d\udfe0', 'medium': '\ud83d\udfe1', 'low': '\ud83d\udfe2'}[severity]\n report += f\"- {emoji} **{severity.upper()}:** {count}\\n\"\n \n if results.get('report_url'):\n report += f\"\\n[View detailed report]({results['report_url']})\\n\"\n \n return report\n```\n\nGenerate and test this plugin:\n\n```bash\n# Generate the manifest\nspaceforge generate security_scanner.py\n\n# Test locally\nexport API_TOKEN=\"your-token\"\nexport SEVERITY_THRESHOLD=\"high\"\nspaceforge runner after_plan\n```\n\n## Speeding up plugin execution\n\nThere are a few things you can do to speed up plugin execution.\n\n1. Ensure your runner has `spaceforge` preinstalled. This will avoid the overhead of installing it during the run. (15-30 seconds)\n2. If youre using binaries, we will only install the binary if its not found. You can gain a few seconds by ensuring its already on the runner.\n3. If your plugin has a lot of dependencies, consider using a prebuilt runner image with your plugin and its dependencies installed. This avoids the overhead of installing them during each run.\n4. Ensure your runner has enough core resources (CPU, memory) to handle the plugin execution efficiently. If your plugin is resource-intensive, consider using a more powerful runner.\n\n## Next Steps\n\n1. **Install spaceforge:** `pip install spaceforge`\n2. **Create your plugin:** Start with the quick start example\n3. **Test locally:** Use the runner command to test your hooks\n4. **Generate manifest:** Use the generate command to create plugin.yaml\n5. **Upload to Spacelift:** Add your plugin manifest to your Spacelift account\n\nFor more advanced examples, see the [plugins](plugins/) directory in this repository.\n",
"bugtrack_url": null,
"license": "MIT",
"summary": "A Python framework for building Spacelift plugins",
"version": "1.1.2",
"project_urls": {
"Bug Reports": "https://github.com/spacelift-io/plugins/issues",
"Documentation": "https://github.com/spacelift-io/plugins#readme",
"Homepage": "https://github.com/spacelift-io/plugins",
"Repository": "https://github.com/spacelift-io/plugins"
},
"split_keywords": [
"spacelift",
" plugin",
" framework",
" infrastructure",
" devops",
" spaceforge"
],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "87123c231b413f07dcf04ec5c2a9b8529d1a64ab926be8055b2af4875b9bde7a",
"md5": "15293d7b9f4e33a7232d821d3b2204a5",
"sha256": "b5db04677fe4a978e88f76cdc70b220ad74b7cdda70a4b312932524fcdf4c895"
},
"downloads": -1,
"filename": "spaceforge-1.1.2-py3-none-any.whl",
"has_sig": false,
"md5_digest": "15293d7b9f4e33a7232d821d3b2204a5",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.9",
"size": 52273,
"upload_time": "2025-08-29T17:17:14",
"upload_time_iso_8601": "2025-08-29T17:17:14.248888Z",
"url": "https://files.pythonhosted.org/packages/87/12/3c231b413f07dcf04ec5c2a9b8529d1a64ab926be8055b2af4875b9bde7a/spaceforge-1.1.2-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": null,
"digests": {
"blake2b_256": "b9a7767245871be47d73c2a17004cbc6be6753ef7cd99ba4def0057ee752dd95",
"md5": "6befbacff6c4cd1a0e00076770e27db0",
"sha256": "fdbef6c9a4c2355a97e03b83c16b98b8ee61bb24e0ad52c7619616a5e7ebdbac"
},
"downloads": -1,
"filename": "spaceforge-1.1.2.tar.gz",
"has_sig": false,
"md5_digest": "6befbacff6c4cd1a0e00076770e27db0",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.9",
"size": 62026,
"upload_time": "2025-08-29T17:17:15",
"upload_time_iso_8601": "2025-08-29T17:17:15.997680Z",
"url": "https://files.pythonhosted.org/packages/b9/a7/767245871be47d73c2a17004cbc6be6753ef7cd99ba4def0057ee752dd95/spaceforge-1.1.2.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-08-29 17:17:15",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "spacelift-io",
"github_project": "plugins",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"lcname": "spaceforge"
}
JSON
