feat: Script Generator Phase 1+2 — backend, engine, API, frontend, template editor, parameter detector

Complete Script Generator feature including:

Backend:
- ScriptCategory, ScriptTemplate, ScriptGeneration models
- ScriptTemplateEngine with substitution, filters, sanitization
- CRUD + share API endpoints with permission checks
- Integration tests for permissions and sharing
- Migration 057 with AD User Management seed templates

Frontend — Script Library:
- Browse templates with category tabs and search
- Configure pane with parameter form and script generation
- Script preview with live substitution and copy/download
- scriptGeneratorStore Zustand store

Frontend — Template Editor:
- Full CRUD form with metadata, script body (Monaco Editor), parameters
- ParameterSchemaBuilder with visual builder + JSON toggle
- ScriptManagePage with routing and nav link

Frontend — Parameter Detector:
- Client-side PowerShell parameter detection engine
- Detects script-level param() blocks and variable assignments
- Type inference from PS type annotations and value patterns
- ParameterDetectorStepper one-by-one review UI with accept/skip

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit was merged in pull request #105.
This commit is contained in:
chihlasm
2026-03-14 20:18:59 -04:00
committed by GitHub
parent 83b13fcd26
commit d4dbf44781
50 changed files with 11916 additions and 11 deletions

View File

@@ -18,6 +18,7 @@ from app.models.survey_response import SurveyResponse
from app.models.survey_invite import SurveyInvite
from app.models.ai_suggestion import AISuggestion # noqa: F401
from app.models.kb_import import KBImport, KBImportNode # noqa: F401
from app.models.script_template import ScriptCategory, ScriptTemplate, ScriptGeneration # noqa: F401
from app.core.config import settings
# this is the Alembic Config object

View File

@@ -0,0 +1,690 @@
"""add script templates
Revision ID: 057
Revises: 056
Create Date: 2026-03-12
"""
from alembic import op
import sqlalchemy as sa
from sqlalchemy.dialects.postgresql import UUID, JSONB, ENUM as PG_ENUM
import uuid
import json
from datetime import datetime, timezone
revision = "057"
down_revision = "056"
branch_labels = None
depends_on = None
def upgrade() -> None:
# ── Create enum ──────────────────────────────────────────────────────
op.execute("CREATE TYPE script_complexity AS ENUM ('beginner', 'intermediate', 'advanced')")
# ── script_categories ────────────────────────────────────────────────
op.create_table(
"script_categories",
sa.Column("id", UUID(as_uuid=True), primary_key=True),
sa.Column("name", sa.String(100), nullable=False),
sa.Column("slug", sa.String(100), nullable=False, unique=True),
sa.Column("description", sa.Text, nullable=True),
sa.Column("icon", sa.String(50), nullable=True),
sa.Column("sort_order", sa.Integer, nullable=False, server_default=sa.text("0")),
sa.Column("is_active", sa.Boolean, nullable=False, server_default=sa.text("true")),
sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
)
op.create_index("ix_script_categories_slug", "script_categories", ["slug"])
# ── script_templates ─────────────────────────────────────────────────
op.create_table(
"script_templates",
sa.Column("id", UUID(as_uuid=True), primary_key=True),
sa.Column("category_id", UUID(as_uuid=True), sa.ForeignKey("script_categories.id", ondelete="RESTRICT"), nullable=False),
sa.Column("team_id", UUID(as_uuid=True), sa.ForeignKey("teams.id", ondelete="CASCADE"), nullable=True),
sa.Column("created_by", UUID(as_uuid=True), sa.ForeignKey("users.id", ondelete="SET NULL"), nullable=True),
sa.Column("name", sa.String(200), nullable=False),
sa.Column("slug", sa.String(200), nullable=False),
sa.Column("description", sa.Text, nullable=True),
sa.Column("use_case", sa.Text, nullable=True),
sa.Column("script_body", sa.Text, nullable=False),
sa.Column("parameters_schema", JSONB, nullable=False, server_default=sa.text("'{}'::jsonb")),
sa.Column("default_values", JSONB, nullable=False, server_default=sa.text("'{}'::jsonb")),
sa.Column("validation_rules", JSONB, nullable=False, server_default=sa.text("'{}'::jsonb")),
sa.Column("tags", JSONB, nullable=False, server_default=sa.text("'[]'::jsonb")),
sa.Column("complexity", PG_ENUM("beginner", "intermediate", "advanced", name="script_complexity", create_type=False), nullable=False, server_default=sa.text("'beginner'")),
sa.Column("estimated_runtime", sa.String(50), nullable=True),
sa.Column("requires_elevation", sa.Boolean, nullable=False, server_default=sa.text("true")),
sa.Column("requires_modules", JSONB, nullable=False, server_default=sa.text("'[]'::jsonb")),
sa.Column("version", sa.Integer, nullable=False, server_default=sa.text("1")),
sa.Column("is_verified", sa.Boolean, nullable=False, server_default=sa.text("false")),
sa.Column("is_active", sa.Boolean, nullable=False, server_default=sa.text("true")),
sa.Column("usage_count", sa.Integer, nullable=False, server_default=sa.text("0")),
sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
)
op.create_index("ix_script_templates_category_id", "script_templates", ["category_id"])
op.create_index("ix_script_templates_team_id", "script_templates", ["team_id"])
op.create_index("ix_script_templates_slug", "script_templates", ["slug"])
# ── script_generations ───────────────────────────────────────────────
op.create_table(
"script_generations",
sa.Column("id", UUID(as_uuid=True), primary_key=True),
sa.Column("template_id", UUID(as_uuid=True), sa.ForeignKey("script_templates.id", ondelete="RESTRICT"), nullable=False),
sa.Column("user_id", UUID(as_uuid=True), sa.ForeignKey("users.id", ondelete="CASCADE"), nullable=False),
sa.Column("team_id", UUID(as_uuid=True), sa.ForeignKey("teams.id", ondelete="SET NULL"), nullable=True),
sa.Column("session_id", UUID(as_uuid=True), sa.ForeignKey("sessions.id", ondelete="SET NULL"), nullable=True),
sa.Column("parameters_used", JSONB, nullable=False, server_default=sa.text("'{}'::jsonb")),
sa.Column("generated_script", sa.Text, nullable=False),
sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
)
op.create_index("ix_script_generations_template_id", "script_generations", ["template_id"])
op.create_index("ix_script_generations_user_id", "script_generations", ["user_id"])
op.create_index("ix_script_generations_session_id", "script_generations", ["session_id"])
# ── Seed: Active Directory category ──────────────────────────────────
now = datetime.now(timezone.utc)
cat_id = uuid.UUID("00000000-0000-0000-0000-000000000001")
conn = op.get_bind()
conn.execute(
sa.text("""
INSERT INTO script_categories (id, name, slug, description, icon, sort_order, is_active, created_at, updated_at)
VALUES (:id, :name, :slug, :description, :icon, :sort_order, true, :now, :now)
"""),
{
"id": cat_id,
"name": "Active Directory",
"slug": "active-directory",
"description": "User account and group management scripts for Active Directory environments",
"icon": "shield-check",
"sort_order": 1,
"now": now,
}
)
templates = _get_seed_templates(cat_id, now)
for tmpl in templates:
conn.execute(
sa.text("""
INSERT INTO script_templates (
id, category_id, name, slug, description, use_case,
script_body, parameters_schema, default_values, validation_rules,
tags, complexity, estimated_runtime, requires_elevation,
requires_modules, version, is_verified, is_active, usage_count,
created_at, updated_at
) VALUES (
:id, :category_id, :name, :slug, :description, :use_case,
:script_body, CAST(:parameters_schema AS jsonb), CAST(:default_values AS jsonb), CAST(:validation_rules AS jsonb),
CAST(:tags AS jsonb), :complexity, :estimated_runtime, :requires_elevation,
CAST(:requires_modules AS jsonb), 1, true, true, 0,
:now, :now
)
"""),
tmpl,
)
def downgrade() -> None:
op.drop_table("script_generations")
op.drop_table("script_templates")
op.drop_table("script_categories")
op.execute("DROP TYPE IF EXISTS script_complexity")
def _get_seed_templates(cat_id: uuid.UUID, now: datetime) -> list[dict]:
"""Return seed data for the six AD User Management templates."""
def tmpl(
id_suffix: int,
name: str,
slug: str,
description: str,
use_case: str,
script_body: str,
parameters_schema: dict,
complexity: str,
estimated_runtime: str,
requires_elevation: bool,
tags: list[str],
) -> dict:
return {
"id": uuid.UUID(f"00000000-0000-0000-0001-{id_suffix:012d}"),
"category_id": cat_id,
"name": name,
"slug": slug,
"description": description,
"use_case": use_case,
"script_body": script_body,
"parameters_schema": json.dumps(parameters_schema),
"default_values": json.dumps({}),
"validation_rules": json.dumps({}),
"tags": json.dumps(tags),
"complexity": complexity,
"estimated_runtime": estimated_runtime,
"requires_elevation": requires_elevation,
"requires_modules": json.dumps([]),
"now": now,
}
return [
# ── 1: Create AD User ─────────────────────────────────────────────
tmpl(
id_suffix=1,
name="Create AD User Account",
slug="create-ad-user",
description="Creates a new Active Directory user account with full property configuration.",
use_case="Use when onboarding a new employee or creating a service account in Active Directory.",
complexity="intermediate",
estimated_runtime="< 5 seconds",
requires_elevation=True,
tags=["active-directory", "user-management", "onboarding"],
parameters_schema={
"parameters": [
{"key": "first_name", "label": "First Name", "type": "text", "required": True, "group": "User Identity", "order": 1, "validation": {"pattern": "^[a-zA-Z\\-']+$", "maxLength": 64}},
{"key": "last_name", "label": "Last Name", "type": "text", "required": True, "group": "User Identity", "order": 2, "validation": {"pattern": "^[a-zA-Z\\-']+$", "maxLength": 64}},
{"key": "sam_account_name", "label": "SAM Account Name", "type": "text", "required": True, "group": "User Identity", "order": 3, "placeholder": "jsmith", "help_text": "The logon name (typically firstname.lastname or first initial + last name). Max 20 characters."},
{"key": "upn_suffix", "label": "UPN Suffix", "type": "text", "required": True, "group": "User Identity", "order": 4, "placeholder": "contoso.com"},
{"key": "ou_path", "label": "Organizational Unit", "type": "text", "required": True, "group": "AD Configuration", "order": 5, "placeholder": "OU=Users,DC=contoso,DC=com", "validation": {"pattern": "^(OU|CN)=.+"}},
{"key": "password", "label": "Initial Password", "type": "password", "required": True, "group": "Security", "order": 6, "sensitive": True},
{"key": "job_title", "label": "Job Title", "type": "text", "required": False, "group": "Profile", "order": 7},
{"key": "department", "label": "Department", "type": "text", "required": False, "group": "Profile", "order": 8},
{"key": "groups", "label": "Security Groups", "type": "multi_text", "required": False, "group": "Group Membership", "order": 9, "placeholder": "Domain Users"},
{"key": "force_password_change", "label": "Force Password Change at Logon", "type": "boolean", "required": False, "group": "Security", "order": 10, "default": True},
{"key": "account_enabled", "label": "Enable Account", "type": "boolean", "required": False, "group": "Security", "order": 11, "default": True},
]
},
script_body=r"""# ============================================================
# Create AD User Account
# Generated by ResolutionFlow Script Generator
# Template Version: 1
# ============================================================
#Requires -Modules ActiveDirectory
#Requires -RunAsAdministrator
$FirstName = '{{ first_name }}'
$LastName = '{{ last_name }}'
$SamAccountName = '{{ sam_account_name }}'
$UPNSuffix = '{{ upn_suffix }}'
$OUPath = '{{ ou_path }}'
$Password = {{ password | as_secure_string }}
$ForcePasswordChange = {{ force_password_change | as_bool }}
$AccountEnabled = {{ account_enabled | as_bool }}
{% if job_title %}
$JobTitle = '{{ job_title }}'
{% endif %}
{% if department %}
$Department = '{{ department }}'
{% endif %}
{% if groups %}
$Groups = @({{ groups | as_array }})
{% endif %}
$DisplayName = "$FirstName $LastName"
$UPN = "$SamAccountName@$UPNSuffix"
try {
Import-Module ActiveDirectory -ErrorAction Stop
Write-Host "[CHECK] ActiveDirectory module loaded" -ForegroundColor Green
} catch {
Write-Host "[ERROR] ActiveDirectory module not available: $_" -ForegroundColor Red
exit 1
}
try {
$userParams = @{
Name = $DisplayName
GivenName = $FirstName
Surname = $LastName
SamAccountName = $SamAccountName
UserPrincipalName = $UPN
Path = $OUPath
AccountPassword = $Password
Enabled = $AccountEnabled
ChangePasswordAtLogon = $ForcePasswordChange
}
{% if job_title %}
$userParams["Title"] = $JobTitle
{% endif %}
{% if department %}
$userParams["Department"] = $Department
{% endif %}
New-ADUser @userParams -ErrorAction Stop
Write-Host "[OK] User account created: $SamAccountName ($DisplayName)" -ForegroundColor Green
} catch {
Write-Host "[ERROR] Failed to create user: $_" -ForegroundColor Red
exit 1
}
{% if groups %}
foreach ($group in $Groups) {
try {
Add-ADGroupMember -Identity $group -Members $SamAccountName -ErrorAction Stop
Write-Host "[OK] Added to group: $group" -ForegroundColor Green
} catch {
Write-Host "[WARN] Could not add to group '$group': $_" -ForegroundColor Yellow
}
}
{% endif %}
Write-Host ""
Write-Host "[DONE] User $SamAccountName created successfully." -ForegroundColor Cyan
Write-Host " UPN: $UPN" -ForegroundColor Cyan
""",
),
# ── 2: Disable AD User ────────────────────────────────────────────
tmpl(
id_suffix=2,
name="Disable AD User Account",
slug="disable-ad-user",
description="Disables an Active Directory user account with optional group removal and OU relocation.",
use_case="Use when offboarding an employee or temporarily suspending access.",
complexity="beginner",
estimated_runtime="< 5 seconds",
requires_elevation=True,
tags=["active-directory", "user-management", "offboarding"],
parameters_schema={
"parameters": [
{"key": "sam_account_name", "label": "SAM Account Name", "type": "text", "required": True, "group": "Target Account", "order": 1, "placeholder": "jsmith"},
{"key": "disable_reason", "label": "Reason", "type": "text", "required": False, "group": "Audit", "order": 2, "placeholder": "Employee offboarding"},
{"key": "remove_groups", "label": "Remove from All Groups", "type": "boolean", "required": False, "group": "Options", "order": 3, "default": False},
{"key": "move_to_disabled_ou", "label": "Move to Disabled OU", "type": "boolean", "required": False, "group": "Options", "order": 4, "default": False},
{"key": "disabled_ou_path", "label": "Disabled OU Path", "type": "text", "required": False, "group": "Options", "order": 5, "placeholder": "OU=Disabled,DC=contoso,DC=com"},
]
},
script_body=r"""# ============================================================
# Disable AD User Account
# Generated by ResolutionFlow Script Generator
# ============================================================
#Requires -Modules ActiveDirectory
#Requires -RunAsAdministrator
$SamAccountName = '{{ sam_account_name }}'
{% if disable_reason %}
$DisableReason = '{{ disable_reason }}'
{% endif %}
$RemoveGroups = {{ remove_groups | as_bool }}
$MoveToDisabled = {{ move_to_disabled_ou | as_bool }}
{% if disabled_ou_path %}
$DisabledOUPath = '{{ disabled_ou_path }}'
{% endif %}
try {
Import-Module ActiveDirectory -ErrorAction Stop
} catch {
Write-Host "[ERROR] ActiveDirectory module not available: $_" -ForegroundColor Red; exit 1
}
try {
$user = Get-ADUser -Identity $SamAccountName -Properties MemberOf -ErrorAction Stop
Write-Host "[CHECK] Found user: $($user.DistinguishedName)" -ForegroundColor Green
} catch {
Write-Host "[ERROR] User not found: $SamAccountName" -ForegroundColor Red; exit 1
}
try {
Disable-ADAccount -Identity $SamAccountName -ErrorAction Stop
Write-Host "[OK] Account disabled: $SamAccountName" -ForegroundColor Green
{% if disable_reason %}
Set-ADUser -Identity $SamAccountName -Description $DisableReason
Write-Host "[OK] Disable reason recorded in Description field" -ForegroundColor Green
{% endif %}
} catch {
Write-Host "[ERROR] Failed to disable account: $_" -ForegroundColor Red; exit 1
}
{% if remove_groups %}
if ($RemoveGroups) {
$groups = $user.MemberOf
foreach ($group in $groups) {
try {
Remove-ADGroupMember -Identity $group -Members $SamAccountName -Confirm:$false -ErrorAction Stop
Write-Host "[OK] Removed from group: $group" -ForegroundColor Green
} catch {
Write-Host "[WARN] Could not remove from group: $group" -ForegroundColor Yellow
}
}
}
{% endif %}
{% if move_to_disabled_ou %}
if ($MoveToDisabled) {
try {
Move-ADObject -Identity $user.DistinguishedName -TargetPath $DisabledOUPath -ErrorAction Stop
Write-Host "[OK] Moved to disabled OU: $DisabledOUPath" -ForegroundColor Green
} catch {
Write-Host "[WARN] Could not move to disabled OU: $_" -ForegroundColor Yellow
}
}
{% endif %}
Write-Host ""
Write-Host "[DONE] Account $SamAccountName has been disabled." -ForegroundColor Cyan
""",
),
# ── 3: Reset AD Password ──────────────────────────────────────────
tmpl(
id_suffix=3,
name="Reset AD Password",
slug="reset-ad-password",
description="Resets an Active Directory user password with optional account unlock.",
use_case="Use when a user has forgotten their password or their account is locked out.",
complexity="beginner",
estimated_runtime="< 5 seconds",
requires_elevation=True,
tags=["active-directory", "password", "account-management"],
parameters_schema={
"parameters": [
{"key": "sam_account_name", "label": "SAM Account Name", "type": "text", "required": True, "group": "Target Account", "order": 1},
{"key": "new_password", "label": "New Password", "type": "password", "required": True, "group": "Security", "order": 2, "sensitive": True},
{"key": "force_change_at_logon", "label": "Force Change at Next Logon", "type": "boolean", "required": False, "group": "Security", "order": 3, "default": True},
{"key": "unlock_account", "label": "Unlock Account if Locked", "type": "boolean", "required": False, "group": "Options", "order": 4, "default": True},
]
},
script_body=r"""# ============================================================
# Reset AD Password
# Generated by ResolutionFlow Script Generator
# ============================================================
#Requires -Modules ActiveDirectory
#Requires -RunAsAdministrator
$SamAccountName = '{{ sam_account_name }}'
$NewPassword = {{ new_password | as_secure_string }}
$ForceChange = {{ force_change_at_logon | as_bool }}
$UnlockAccount = {{ unlock_account | as_bool }}
try { Import-Module ActiveDirectory -ErrorAction Stop } catch { Write-Host "[ERROR] $_ " -ForegroundColor Red; exit 1 }
try {
Set-ADAccountPassword -Identity $SamAccountName -NewPassword $NewPassword -Reset -ErrorAction Stop
Write-Host "[OK] Password reset for: $SamAccountName" -ForegroundColor Green
} catch {
Write-Host "[ERROR] Failed to reset password: $_" -ForegroundColor Red; exit 1
}
if ($ForceChange) {
Set-ADUser -Identity $SamAccountName -ChangePasswordAtLogon $true
Write-Host "[OK] User must change password at next logon" -ForegroundColor Green
}
if ($UnlockAccount) {
try {
Unlock-ADAccount -Identity $SamAccountName -ErrorAction Stop
Write-Host "[OK] Account unlocked" -ForegroundColor Green
} catch {
Write-Host "[WARN] Could not unlock account (may not have been locked): $_" -ForegroundColor Yellow
}
}
Write-Host ""
Write-Host "[DONE] Password reset complete for $SamAccountName." -ForegroundColor Cyan
""",
),
# ── 4: Unlock AD Account ──────────────────────────────────────────
tmpl(
id_suffix=4,
name="Unlock AD Account",
slug="unlock-ad-account",
description="Unlocks a locked-out Active Directory user account and optionally shows lockout information.",
use_case="Use when a user is locked out after too many failed login attempts.",
complexity="beginner",
estimated_runtime="< 5 seconds",
requires_elevation=True,
tags=["active-directory", "account-management", "lockout"],
parameters_schema={
"parameters": [
{"key": "sam_account_name", "label": "SAM Account Name", "type": "text", "required": True, "group": "Target Account", "order": 1},
{"key": "show_lockout_info", "label": "Show Lockout Source Info", "type": "boolean", "required": False, "group": "Options", "order": 2, "default": False, "help_text": "Queries domain controllers for lockout source (may be slow)"},
]
},
script_body=r"""# ============================================================
# Unlock AD Account
# Generated by ResolutionFlow Script Generator
# ============================================================
#Requires -Modules ActiveDirectory
#Requires -RunAsAdministrator
$SamAccountName = '{{ sam_account_name }}'
$ShowLockoutInfo = {{ show_lockout_info | as_bool }}
try { Import-Module ActiveDirectory -ErrorAction Stop } catch { Write-Host "[ERROR] $_ " -ForegroundColor Red; exit 1 }
try {
$user = Get-ADUser -Identity $SamAccountName -Properties LockedOut, BadLogonCount, LastBadPasswordAttempt -ErrorAction Stop
Write-Host "[CHECK] User found: $($user.DistinguishedName)" -ForegroundColor Green
Write-Host " Locked Out: $($user.LockedOut)" -ForegroundColor Cyan
Write-Host " Bad Logon Count: $($user.BadLogonCount)" -ForegroundColor Cyan
Write-Host " Last Bad Password Attempt: $($user.LastBadPasswordAttempt)" -ForegroundColor Cyan
} catch {
Write-Host "[ERROR] User not found: $SamAccountName" -ForegroundColor Red; exit 1
}
if (-not $user.LockedOut) {
Write-Host "[INFO] Account is not currently locked." -ForegroundColor Yellow
} else {
try {
Unlock-ADAccount -Identity $SamAccountName -ErrorAction Stop
Write-Host "[OK] Account unlocked: $SamAccountName" -ForegroundColor Green
} catch {
Write-Host "[ERROR] Failed to unlock account: $_" -ForegroundColor Red; exit 1
}
}
{% if show_lockout_info %}
if ($ShowLockoutInfo) {
Write-Host ""
Write-Host "[INFO] Querying domain controllers for lockout source..." -ForegroundColor Cyan
try {
$DCs = Get-ADDomainController -Filter * | Select-Object -ExpandProperty HostName
foreach ($DC in $DCs) {
$events = Get-WinEvent -ComputerName $DC -FilterHashtable @{
LogName = "Security"
Id = 4740
} -MaxEvents 10 -ErrorAction SilentlyContinue |
Where-Object { $_.Properties[0].Value -eq $SamAccountName }
foreach ($e in $events) {
Write-Host " DC: $DC | Source: $($e.Properties[1].Value) | Time: $($e.TimeCreated)" -ForegroundColor Yellow
}
}
} catch {
Write-Host "[WARN] Could not query lockout source: $_" -ForegroundColor Yellow
}
}
{% endif %}
Write-Host ""
Write-Host "[DONE] Unlock operation complete for $SamAccountName." -ForegroundColor Cyan
""",
),
# ── 5: Delete AD User ─────────────────────────────────────────────
tmpl(
id_suffix=5,
name="Delete AD User Account",
slug="delete-ad-user",
description="Permanently deletes an Active Directory user account with optional CSV backup of user properties.",
use_case="Use when permanently removing an account that is no longer needed. Prefer Disable for departing employees.",
complexity="advanced",
estimated_runtime="< 10 seconds",
requires_elevation=True,
tags=["active-directory", "user-management", "destructive"],
parameters_schema={
"parameters": [
{"key": "sam_account_name", "label": "SAM Account Name", "type": "text", "required": True, "group": "Target Account", "order": 1},
{"key": "confirm_deletion", "label": "I confirm this account should be permanently deleted", "type": "boolean", "required": True, "group": "Confirmation", "order": 2, "default": False},
{"key": "backup_to_csv", "label": "Back Up User Properties to CSV First", "type": "boolean", "required": False, "group": "Options", "order": 3, "default": True},
{"key": "backup_path", "label": "CSV Backup Path", "type": "text", "required": False, "group": "Options", "order": 4, "placeholder": "C:\\Backups\\ad-user-backup.csv"},
]
},
script_body=r"""# ============================================================
# Delete AD User Account
# Generated by ResolutionFlow Script Generator
# DESTRUCTIVE OPERATION
# ============================================================
#Requires -Modules ActiveDirectory
#Requires -RunAsAdministrator
$SamAccountName = '{{ sam_account_name }}'
$ConfirmDeletion = {{ confirm_deletion | as_bool }}
$BackupToCSV = {{ backup_to_csv | as_bool }}
{% if backup_path %}
$BackupPath = '{{ backup_path }}'
{% else %}
$BackupPath = "C:\Temp\ad-user-backup-$SamAccountName-$(Get-Date -Format 'yyyyMMdd-HHmmss').csv"
{% endif %}
try { Import-Module ActiveDirectory -ErrorAction Stop } catch { Write-Host "[ERROR] $_ " -ForegroundColor Red; exit 1 }
if (-not $ConfirmDeletion) {
Write-Host "[ABORT] Deletion not confirmed. Set confirm_deletion to true to proceed." -ForegroundColor Red
exit 1
}
try {
$user = Get-ADUser -Identity $SamAccountName -Properties * -ErrorAction Stop
Write-Host "[CHECK] Found user: $($user.DistinguishedName)" -ForegroundColor Yellow
Write-Host "[WARN] About to permanently delete this account." -ForegroundColor Yellow
} catch {
Write-Host "[ERROR] User not found: $SamAccountName" -ForegroundColor Red; exit 1
}
if ($BackupToCSV) {
try {
$user | Select-Object Name, SamAccountName, UserPrincipalName, DistinguishedName, EmailAddress, Title, Department, Enabled, Created, Modified |
Export-Csv -Path $BackupPath -NoTypeInformation -ErrorAction Stop
Write-Host "[OK] User properties backed up to: $BackupPath" -ForegroundColor Green
} catch {
Write-Host "[WARN] Could not write backup: $_" -ForegroundColor Yellow
}
}
try {
Remove-ADUser -Identity $SamAccountName -Confirm:$false -ErrorAction Stop
Write-Host "[OK] User account permanently deleted: $SamAccountName" -ForegroundColor Green
} catch {
Write-Host "[ERROR] Failed to delete user: $_" -ForegroundColor Red; exit 1
}
Write-Host ""
Write-Host "[DONE] Account $SamAccountName has been permanently deleted." -ForegroundColor Cyan
""",
),
# ── 6: Bulk User Import ───────────────────────────────────────────
tmpl(
id_suffix=6,
name="Bulk User Import from CSV",
slug="bulk-user-import",
description="Imports multiple Active Directory user accounts from a CSV file with dry-run support and detailed logging.",
use_case="Use when onboarding multiple employees at once from an HR system export.",
complexity="advanced",
estimated_runtime="1-2 minutes",
requires_elevation=True,
tags=["active-directory", "bulk", "onboarding", "csv"],
parameters_schema={
"parameters": [
{"key": "csv_path", "label": "CSV File Path", "type": "text", "required": True, "group": "Source", "order": 1, "placeholder": "C:\\Imports\\new-users.csv", "help_text": "CSV must have columns: FirstName, LastName, SamAccountName, UPN, Department, JobTitle"},
{"key": "ou_path", "label": "Target Organizational Unit", "type": "text", "required": True, "group": "AD Configuration", "order": 2, "placeholder": "OU=Users,DC=contoso,DC=com"},
{"key": "default_password", "label": "Default Password", "type": "password", "required": True, "group": "Security", "order": 3, "sensitive": True},
{"key": "groups", "label": "Add All Users to Groups", "type": "multi_text", "required": False, "group": "Group Membership", "order": 4},
{"key": "force_password_change", "label": "Force Password Change at Logon", "type": "boolean", "required": False, "group": "Security", "order": 5, "default": True},
{"key": "log_path", "label": "Log File Path", "type": "text", "required": False, "group": "Options", "order": 6, "placeholder": "C:\\Logs\\bulk-import.log"},
{"key": "dry_run", "label": "Dry Run (Preview Only - No Changes Made)", "type": "boolean", "required": False, "group": "Options", "order": 7, "default": False},
]
},
script_body=r"""# ============================================================
# Bulk User Import from CSV
# Generated by ResolutionFlow Script Generator
# ============================================================
# CSV Format: FirstName, LastName, SamAccountName, UPN, Department, JobTitle
# ============================================================
#Requires -Modules ActiveDirectory
#Requires -RunAsAdministrator
$CSVPath = '{{ csv_path }}'
$OUPath = '{{ ou_path }}'
$DefaultPassword = {{ default_password | as_secure_string }}
$ForceChange = {{ force_password_change | as_bool }}
$DryRun = {{ dry_run | as_bool }}
{% if log_path %}
$LogPath = '{{ log_path }}'
{% else %}
$LogPath = "C:\Temp\bulk-import-$(Get-Date -Format 'yyyyMMdd-HHmmss').log"
{% endif %}
{% if groups %}
$DefaultGroups = @({{ groups | as_array }})
{% endif %}
function Write-Log {
param($Message, $Level = "INFO")
$line = "$(Get-Date -Format 'yyyy-MM-dd HH:mm:ss') [$Level] $Message"
Add-Content -Path $LogPath -Value $line -ErrorAction SilentlyContinue
$color = switch ($Level) { "ERROR" { "Red" } "WARN" { "Yellow" } "OK" { "Green" } default { "White" } }
Write-Host $line -ForegroundColor $color
}
try { Import-Module ActiveDirectory -ErrorAction Stop } catch { Write-Log "ActiveDirectory module not available: $_" "ERROR"; exit 1 }
if (-not (Test-Path $CSVPath)) { Write-Log "CSV file not found: $CSVPath" "ERROR"; exit 1 }
$users = Import-Csv -Path $CSVPath
Write-Log "Loaded $($users.Count) users from CSV"
if ($DryRun) { Write-Log "DRY RUN MODE - no changes will be made" "WARN" }
$successCount = 0
$errorCount = 0
foreach ($row in $users) {
$sam = $row.SamAccountName
try {
$existing = Get-ADUser -Filter "SamAccountName -eq '$sam'" -ErrorAction SilentlyContinue
if ($existing) { Write-Log "SKIP: $sam already exists" "WARN"; continue }
if (-not $DryRun) {
New-ADUser `
-Name "$($row.FirstName) $($row.LastName)" `
-GivenName $row.FirstName `
-Surname $row.LastName `
-SamAccountName $sam `
-UserPrincipalName $row.UPN `
-Path $OUPath `
-AccountPassword $DefaultPassword `
-Enabled $true `
-ChangePasswordAtLogon $ForceChange `
-Title $row.JobTitle `
-Department $row.Department `
-ErrorAction Stop
{% if groups %}
foreach ($group in $DefaultGroups) {
try { Add-ADGroupMember -Identity $group -Members $sam -ErrorAction Stop }
catch { Write-Log "WARN: $sam - could not add to group '$group'" "WARN" }
}
{% endif %}
}
Write-Log "$(if ($DryRun) { 'PREVIEW' } else { 'CREATED' }): $sam ($($row.FirstName) $($row.LastName))" "OK"
$successCount++
} catch {
Write-Log "FAILED: $sam - $_" "ERROR"
$errorCount++
}
}
Write-Log ""
Write-Log "[DONE] Import complete. Success: $successCount | Errors: $errorCount | Log: $LogPath" "INFO"
""",
),
]

View File

@@ -0,0 +1,413 @@
"""Script Generator API endpoints."""
from typing import Annotated, Optional
from uuid import UUID
import re
from fastapi import APIRouter, Depends, HTTPException, Query, status
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select, func, or_
from app.core.database import get_db
from app.api.deps import get_current_active_user
from app.core.permissions import can_manage_script_template, can_create_content
from app.models.user import User
from app.models.script_template import ScriptCategory, ScriptTemplate, ScriptGeneration
from app.schemas.script_template import (
ScriptCategoryResponse,
ScriptTemplateCreate,
ScriptTemplateUpdate,
ScriptTemplateListItem,
ScriptTemplateDetail,
ScriptGenerateRequest,
ScriptGenerateResponse,
ScriptGenerationRecord,
)
from app.services.script_template_engine import ScriptTemplateEngine, ScriptRenderError
router = APIRouter(prefix="/scripts", tags=["scripts"])
_engine = ScriptTemplateEngine()
# ── Categories ────────────────────────────────────────────────────────────
@router.get("/categories", response_model=list[ScriptCategoryResponse])
async def list_categories(
db: Annotated[AsyncSession, Depends(get_db)],
current_user: Annotated[User, Depends(get_current_active_user)],
) -> list[ScriptCategoryResponse]:
result = await db.execute(
select(ScriptCategory)
.where(ScriptCategory.is_active == True) # noqa: E712
.order_by(ScriptCategory.sort_order)
)
categories = result.scalars().all()
count_result = await db.execute(
select(ScriptTemplate.category_id, func.count(ScriptTemplate.id))
.where(ScriptTemplate.is_active == True) # noqa: E712
.group_by(ScriptTemplate.category_id)
)
counts = dict(count_result.all())
return [
ScriptCategoryResponse(
id=cat.id,
name=cat.name,
slug=cat.slug,
description=cat.description,
icon=cat.icon,
sort_order=cat.sort_order,
template_count=counts.get(cat.id, 0),
)
for cat in categories
]
# ── Templates ─────────────────────────────────────────────────────────────
@router.get("/templates", response_model=list[ScriptTemplateListItem])
async def list_templates(
db: Annotated[AsyncSession, Depends(get_db)],
current_user: Annotated[User, Depends(get_current_active_user)],
category_slug: Optional[str] = Query(None),
search: Optional[str] = Query(None),
tags: Optional[str] = Query(None, description="Comma-separated tags"),
managed: Optional[bool] = Query(None, description="If true, return only templates this user can edit"),
) -> list[ScriptTemplateListItem]:
query = (
select(ScriptTemplate)
.join(ScriptCategory, ScriptTemplate.category_id == ScriptCategory.id)
.where(ScriptTemplate.is_active == True) # noqa: E712
.where(
or_(
ScriptTemplate.team_id == None, # noqa: E711
ScriptTemplate.team_id == current_user.team_id,
)
)
)
if category_slug:
query = query.where(ScriptCategory.slug == category_slug)
if search:
term = f"%{search.lower()}%"
query = query.where(
or_(
func.lower(ScriptTemplate.name).like(term),
func.lower(ScriptTemplate.description).like(term),
func.lower(ScriptTemplate.slug).like(term),
)
)
if managed:
if current_user.is_super_admin:
pass # super admin can edit all
elif current_user.account_role == "owner":
query = query.where(
or_(
ScriptTemplate.created_by == current_user.id,
ScriptTemplate.team_id != None, # noqa: E711
)
)
else:
# engineers see only their own
query = query.where(ScriptTemplate.created_by == current_user.id)
result = await db.execute(query.order_by(ScriptTemplate.name))
templates = result.scalars().all()
if tags:
tag_list = [t.strip().lower() for t in tags.split(",")]
templates = [
t
for t in templates
if any(tag in [tg.lower() for tg in (t.tags or [])] for tag in tag_list)
]
return [ScriptTemplateListItem.model_validate(t) for t in templates]
@router.get("/templates/{template_id}", response_model=ScriptTemplateDetail)
async def get_template(
template_id: UUID,
db: Annotated[AsyncSession, Depends(get_db)],
current_user: Annotated[User, Depends(get_current_active_user)],
) -> ScriptTemplateDetail:
result = await db.execute(
select(ScriptTemplate).where(
ScriptTemplate.id == template_id,
ScriptTemplate.is_active == True, # noqa: E712
or_(
ScriptTemplate.team_id == None, # noqa: E711
ScriptTemplate.team_id == current_user.team_id,
),
)
)
template = result.scalar_one_or_none()
if not template:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND, detail="Template not found"
)
return ScriptTemplateDetail.model_validate(template)
@router.post(
"/templates",
response_model=ScriptTemplateDetail,
status_code=status.HTTP_201_CREATED,
)
async def create_template(
data: ScriptTemplateCreate,
db: Annotated[AsyncSession, Depends(get_db)],
current_user: Annotated[User, Depends(get_current_active_user)],
) -> ScriptTemplateDetail:
if not can_create_content(current_user):
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Engineer access required to create templates",
)
cat_result = await db.execute(
select(ScriptCategory).where(
ScriptCategory.id == data.category_id,
ScriptCategory.is_active == True, # noqa: E712
)
)
if not cat_result.scalar_one_or_none():
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND, detail="Category not found"
)
slug = re.sub(r"[^a-z0-9]+", "-", data.name.lower()).strip("-")
template = ScriptTemplate(
category_id=data.category_id,
team_id=current_user.team_id,
created_by=current_user.id,
name=data.name,
slug=slug,
description=data.description,
use_case=data.use_case,
script_body=data.script_body,
parameters_schema=data.parameters_schema,
default_values=data.default_values,
validation_rules=data.validation_rules,
tags=data.tags,
complexity=data.complexity,
estimated_runtime=data.estimated_runtime,
requires_elevation=data.requires_elevation,
requires_modules=data.requires_modules,
)
db.add(template)
await db.commit()
await db.refresh(template)
return ScriptTemplateDetail.model_validate(template)
@router.put("/templates/{template_id}", response_model=ScriptTemplateDetail)
async def update_template(
template_id: UUID,
data: ScriptTemplateUpdate,
db: Annotated[AsyncSession, Depends(get_db)],
current_user: Annotated[User, Depends(get_current_active_user)],
) -> ScriptTemplateDetail:
result = await db.execute(
select(ScriptTemplate).where(
ScriptTemplate.id == template_id,
ScriptTemplate.is_active == True, # noqa: E712
)
)
template = result.scalar_one_or_none()
if not template:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Template not found",
)
if not can_manage_script_template(current_user, template.created_by, template.team_id):
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="You don't have permission to edit this template",
)
update_data = data.model_dump(exclude_unset=True)
if "script_body" in update_data or "parameters_schema" in update_data:
template.version += 1
for field, value in update_data.items():
setattr(template, field, value)
await db.commit()
await db.refresh(template)
return ScriptTemplateDetail.model_validate(template)
@router.delete("/templates/{template_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_template(
template_id: UUID,
db: Annotated[AsyncSession, Depends(get_db)],
current_user: Annotated[User, Depends(get_current_active_user)],
) -> None:
result = await db.execute(
select(ScriptTemplate).where(
ScriptTemplate.id == template_id,
ScriptTemplate.is_active == True, # noqa: E712
)
)
template = result.scalar_one_or_none()
if not template:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Template not found",
)
if not can_manage_script_template(current_user, template.created_by, template.team_id):
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="You don't have permission to delete this template",
)
template.is_active = False
await db.commit()
@router.patch("/templates/{template_id}/share", response_model=ScriptTemplateDetail)
async def share_template(
template_id: UUID,
db: Annotated[AsyncSession, Depends(get_db)],
current_user: Annotated[User, Depends(get_current_active_user)],
shared: bool = Query(..., description="true to share with team, false to make personal"),
) -> ScriptTemplateDetail:
"""Toggle team sharing for a template. Owner/admin/super_admin only."""
if not (current_user.is_super_admin or current_user.account_role == "owner"):
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Only account owners and admins can share templates",
)
result = await db.execute(
select(ScriptTemplate).where(
ScriptTemplate.id == template_id,
ScriptTemplate.is_active == True, # noqa: E712
)
)
template = result.scalar_one_or_none()
if not template:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Template not found",
)
if shared:
template.team_id = current_user.team_id
else:
template.team_id = None
await db.commit()
await db.refresh(template)
return ScriptTemplateDetail.model_validate(template)
# ── Generate ──────────────────────────────────────────────────────────────
@router.post("/generate", response_model=ScriptGenerateResponse)
async def generate_script(
data: ScriptGenerateRequest,
db: Annotated[AsyncSession, Depends(get_db)],
current_user: Annotated[User, Depends(get_current_active_user)],
) -> ScriptGenerateResponse:
result = await db.execute(
select(ScriptTemplate).where(
ScriptTemplate.id == data.template_id,
ScriptTemplate.is_active == True, # noqa: E712
or_(
ScriptTemplate.team_id == None, # noqa: E711
ScriptTemplate.team_id == current_user.team_id,
),
)
)
template = result.scalar_one_or_none()
if not template:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND, detail="Template not found"
)
try:
rendered_script = _engine.render(template.script_body, data.parameters)
except ScriptRenderError as e:
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=str(e)
)
params_schema = template.parameters_schema or {}
sensitive_keys = {
p["key"]
for p in params_schema.get("parameters", [])
if p.get("sensitive", False)
}
redacted_params = _engine.redact_sensitive(data.parameters, sensitive_keys)
generation = ScriptGeneration(
template_id=template.id,
user_id=current_user.id,
team_id=current_user.team_id,
session_id=data.session_id,
parameters_used=redacted_params,
generated_script=rendered_script,
)
db.add(generation)
template.usage_count += 1
await db.commit()
await db.refresh(generation)
warnings: list[str] = []
if template.requires_elevation:
warnings.append("This script requires 'Run as Administrator'")
return ScriptGenerateResponse(
id=generation.id,
script=rendered_script,
warnings=warnings,
metadata={
"template_name": template.name,
"template_version": template.version,
"requires_elevation": template.requires_elevation,
"requires_modules": template.requires_modules,
"generated_at": generation.created_at.isoformat(),
"estimated_runtime": template.estimated_runtime,
},
)
# ── Generations history ───────────────────────────────────────────────────
@router.get("/generations", response_model=list[ScriptGenerationRecord])
async def list_generations(
db: Annotated[AsyncSession, Depends(get_db)],
current_user: Annotated[User, Depends(get_current_active_user)],
limit: int = Query(20, ge=1, le=100),
offset: int = Query(0, ge=0),
) -> list[ScriptGenerationRecord]:
result = await db.execute(
select(ScriptGeneration, ScriptTemplate.name)
.join(ScriptTemplate, ScriptGeneration.template_id == ScriptTemplate.id)
.where(ScriptGeneration.user_id == current_user.id)
.order_by(ScriptGeneration.created_at.desc())
.limit(limit)
.offset(offset)
)
rows = result.all()
return [
ScriptGenerationRecord(
id=gen.id,
template_id=gen.template_id,
template_name=name,
parameters_used=gen.parameters_used,
created_at=gen.created_at,
)
for gen, name in rows
]

View File

@@ -16,6 +16,7 @@ from app.api.endpoints import tree_transfer
from app.api.endpoints import ai_suggestions
from app.api.endpoints import kb_accelerator
from app.api.endpoints import beta_signup
from app.api.endpoints import scripts
api_router = APIRouter()
@@ -56,3 +57,4 @@ api_router.include_router(tree_transfer.router)
api_router.include_router(ai_suggestions.router)
api_router.include_router(kb_accelerator.router)
api_router.include_router(beta_signup.router)
api_router.include_router(scripts.router)

View File

@@ -169,3 +169,19 @@ def can_create_step_category(user: User, account_id: Optional[UUID]) -> bool:
if user.account_role == "owner" and account_id == user.account_id and user.account_id is not None:
return True
return False
def can_manage_script_template(user: User, template_created_by: Optional[UUID], template_account_id: Optional[UUID] = None) -> bool:
"""Can the user edit/delete this script template?
- Super admins can manage any template
- Account owners can manage any template in their account
- Engineers can manage templates they created
"""
if user.is_super_admin:
return True
if user.account_role == "owner" and template_account_id == user.account_id and user.account_id is not None:
return True
if template_created_by == user.id:
return True
return False

View File

@@ -35,6 +35,7 @@ from .assistant_chat import AssistantChat
from .survey_response import SurveyResponse
from .survey_invite import SurveyInvite
from .kb_import import KBImport, KBImportNode
from .script_template import ScriptCategory, ScriptTemplate, ScriptGeneration
__all__ = [
"User",
@@ -82,4 +83,7 @@ __all__ = [
"SurveyInvite",
"KBImport",
"KBImportNode",
"ScriptCategory",
"ScriptTemplate",
"ScriptGeneration",
]

View File

@@ -0,0 +1,107 @@
import uuid
from datetime import datetime, timezone
from typing import Optional, TYPE_CHECKING
from sqlalchemy import String, Text, DateTime, ForeignKey, Boolean, Integer, Enum as SAEnum
from sqlalchemy.orm import Mapped, mapped_column, relationship
from sqlalchemy.dialects.postgresql import UUID, JSONB
from app.core.database import Base
if TYPE_CHECKING:
from app.models.user import User
from app.models.team import Team
from app.models.session import Session
class ScriptCategory(Base):
__tablename__ = "script_categories"
id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
name: Mapped[str] = mapped_column(String(100), nullable=False)
slug: Mapped[str] = mapped_column(String(100), nullable=False, unique=True, index=True)
description: Mapped[Optional[str]] = mapped_column(Text, nullable=True)
icon: Mapped[Optional[str]] = mapped_column(String(50), nullable=True)
sort_order: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
is_active: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), default=lambda: datetime.now(timezone.utc)
)
updated_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
default=lambda: datetime.now(timezone.utc),
onupdate=lambda: datetime.now(timezone.utc),
)
templates: Mapped[list["ScriptTemplate"]] = relationship("ScriptTemplate", back_populates="category")
class ScriptTemplate(Base):
__tablename__ = "script_templates"
id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
category_id: Mapped[uuid.UUID] = mapped_column(
UUID(as_uuid=True), ForeignKey("script_categories.id", ondelete="RESTRICT"), nullable=False, index=True
)
team_id: Mapped[Optional[uuid.UUID]] = mapped_column(
UUID(as_uuid=True), ForeignKey("teams.id", ondelete="CASCADE"), nullable=True, index=True
)
created_by: Mapped[Optional[uuid.UUID]] = mapped_column(
UUID(as_uuid=True), ForeignKey("users.id", ondelete="SET NULL"), nullable=True
)
name: Mapped[str] = mapped_column(String(200), nullable=False)
slug: Mapped[str] = mapped_column(String(200), nullable=False, index=True)
description: Mapped[Optional[str]] = mapped_column(Text, nullable=True)
use_case: Mapped[Optional[str]] = mapped_column(Text, nullable=True)
script_body: Mapped[str] = mapped_column(Text, nullable=False)
parameters_schema: Mapped[dict] = mapped_column(JSONB, nullable=False, default=dict)
default_values: Mapped[dict] = mapped_column(JSONB, nullable=False, default=dict)
validation_rules: Mapped[dict] = mapped_column(JSONB, nullable=False, default=dict)
tags: Mapped[list] = mapped_column(JSONB, nullable=False, default=list)
complexity: Mapped[str] = mapped_column(
SAEnum("beginner", "intermediate", "advanced", name="script_complexity"), nullable=False, default="beginner"
)
estimated_runtime: Mapped[Optional[str]] = mapped_column(String(50), nullable=True)
requires_elevation: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
requires_modules: Mapped[list] = mapped_column(JSONB, nullable=False, default=list)
version: Mapped[int] = mapped_column(Integer, nullable=False, default=1)
is_verified: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
is_active: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True)
usage_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), default=lambda: datetime.now(timezone.utc)
)
updated_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
default=lambda: datetime.now(timezone.utc),
onupdate=lambda: datetime.now(timezone.utc),
)
category: Mapped["ScriptCategory"] = relationship("ScriptCategory", back_populates="templates")
team: Mapped[Optional["Team"]] = relationship("Team")
creator: Mapped[Optional["User"]] = relationship("User", foreign_keys=[created_by])
generations: Mapped[list["ScriptGeneration"]] = relationship("ScriptGeneration", back_populates="template")
class ScriptGeneration(Base):
__tablename__ = "script_generations"
id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
template_id: Mapped[uuid.UUID] = mapped_column(
UUID(as_uuid=True), ForeignKey("script_templates.id", ondelete="RESTRICT"), nullable=False, index=True
)
user_id: Mapped[uuid.UUID] = mapped_column(
UUID(as_uuid=True), ForeignKey("users.id", ondelete="CASCADE"), nullable=False, index=True
)
team_id: Mapped[Optional[uuid.UUID]] = mapped_column(
UUID(as_uuid=True), ForeignKey("teams.id", ondelete="SET NULL"), nullable=True, index=True
)
session_id: Mapped[Optional[uuid.UUID]] = mapped_column(
UUID(as_uuid=True), ForeignKey("sessions.id", ondelete="SET NULL"), nullable=True, index=True
)
parameters_used: Mapped[dict] = mapped_column(JSONB, nullable=False, default=dict)
generated_script: Mapped[str] = mapped_column(Text, nullable=False)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), default=lambda: datetime.now(timezone.utc)
)
template: Mapped["ScriptTemplate"] = relationship("ScriptTemplate", back_populates="generations")
user: Mapped["User"] = relationship("User")

View File

@@ -10,6 +10,11 @@ from .ai_builder import (
AIStartResponse, AIScaffoldResponse, AIBranchDetailResponse, AIAssembleResponse,
AIQuotaStatusResponse,
)
from .script_template import (
ScriptCategoryResponse,
ScriptTemplateCreate, ScriptTemplateUpdate, ScriptTemplateListItem, ScriptTemplateDetail,
ScriptGenerateRequest, ScriptGenerateResponse, ScriptGenerationRecord,
)
__all__ = [
# User
@@ -30,4 +35,8 @@ __all__ = [
"AIStartRequest", "AIScaffoldRequest", "AIBranchDetailRequest", "AIAssembleRequest",
"AIStartResponse", "AIScaffoldResponse", "AIBranchDetailResponse", "AIAssembleResponse",
"AIQuotaStatusResponse",
# Script Generator
"ScriptCategoryResponse",
"ScriptTemplateCreate", "ScriptTemplateUpdate", "ScriptTemplateListItem", "ScriptTemplateDetail",
"ScriptGenerateRequest", "ScriptGenerateResponse", "ScriptGenerationRecord",
]

View File

@@ -0,0 +1,138 @@
from datetime import datetime
from typing import Optional, Any
from uuid import UUID
from pydantic import BaseModel, Field
# ── Parameter schema types ──────────────────────────────────────────────────
class ScriptParameterValidation(BaseModel):
pattern: Optional[str] = None
min_length: Optional[int] = None
max_length: Optional[int] = None
min_value: Optional[float] = None
max_value: Optional[float] = None
class ScriptParameter(BaseModel):
key: str
label: str
type: str # text | password | select | boolean | multi_text | number | textarea
required: bool = True
placeholder: Optional[str] = None
group: Optional[str] = None
order: int = 0
help_text: Optional[str] = None
options: Optional[list[dict]] = None # for select type: [{value, label}]
default: Optional[Any] = None
validation: Optional[ScriptParameterValidation] = None
sensitive: bool = False # password fields → redacted in generation record
class ScriptParametersSchema(BaseModel):
parameters: list[ScriptParameter]
# ── Category ────────────────────────────────────────────────────────────────
class ScriptCategoryResponse(BaseModel):
id: UUID
name: str
slug: str
description: Optional[str] = None
icon: Optional[str] = None
sort_order: int
template_count: int = 0
class Config:
from_attributes = True
# ── Template ────────────────────────────────────────────────────────────────
class ScriptTemplateCreate(BaseModel):
category_id: UUID
name: str = Field(..., min_length=1, max_length=200)
description: Optional[str] = None
use_case: Optional[str] = None
script_body: str = Field(..., min_length=1)
parameters_schema: dict = Field(default_factory=dict)
default_values: dict = Field(default_factory=dict)
validation_rules: dict = Field(default_factory=dict)
tags: list[str] = Field(default_factory=list)
complexity: str = Field(default="beginner", pattern="^(beginner|intermediate|advanced)$")
estimated_runtime: Optional[str] = None
requires_elevation: bool = False
requires_modules: list[str] = Field(default_factory=list)
class ScriptTemplateUpdate(BaseModel):
name: Optional[str] = Field(None, min_length=1, max_length=200)
description: Optional[str] = None
use_case: Optional[str] = None
script_body: Optional[str] = None
parameters_schema: Optional[dict] = None
default_values: Optional[dict] = None
validation_rules: Optional[dict] = None
tags: Optional[list[str]] = None
complexity: Optional[str] = Field(None, pattern="^(beginner|intermediate|advanced)$")
estimated_runtime: Optional[str] = None
requires_elevation: Optional[bool] = None
requires_modules: Optional[list[str]] = None
is_active: Optional[bool] = None
class ScriptTemplateListItem(BaseModel):
id: UUID
category_id: UUID
team_id: Optional[UUID] = None
created_by: Optional[UUID] = None
name: str
slug: str
description: Optional[str] = None
tags: list[str]
complexity: str
estimated_runtime: Optional[str] = None
requires_elevation: bool
requires_modules: list[str]
is_verified: bool
usage_count: int
class Config:
from_attributes = True
class ScriptTemplateDetail(ScriptTemplateListItem):
use_case: Optional[str] = None
script_body: str
parameters_schema: dict
default_values: dict
validation_rules: dict
version: int
created_at: datetime
updated_at: datetime
class Config:
from_attributes = True
# ── Generation ───────────────────────────────────────────────────────────────
class ScriptGenerateRequest(BaseModel):
template_id: UUID
parameters: dict[str, Any]
session_id: Optional[UUID] = None
class ScriptGenerateResponse(BaseModel):
id: UUID
script: str
warnings: list[str] = Field(default_factory=list)
metadata: dict
class Config:
from_attributes = True
class ScriptGenerationRecord(BaseModel):
id: UUID
template_id: UUID
template_name: str
parameters_used: dict # passwords already redacted
created_at: datetime
class Config:
from_attributes = True

View File

@@ -0,0 +1,139 @@
"""
ScriptTemplateEngine — renders parameterized PowerShell templates.
Template syntax:
{{ param_name }} — simple substitution (string-escaped)
{{ param | filter }} — substitution with filter applied
{% if param %} ... {% endif %} — conditional block
Security: all string values are PowerShell-escaped before insertion.
Sensitive parameter values are never stored in generation records — use redact_sensitive().
"""
import re
from typing import Any
class ScriptRenderError(Exception):
"""Raised when a required parameter is missing or rendering fails."""
class ScriptTemplateEngine:
# ── Public API ────────────────────────────────────────────────────────
def render(self, body: str, parameters: dict[str, Any]) -> str:
"""Render a template body with the given parameters. Raises ScriptRenderError on missing required params."""
result = self._process_conditionals(body, parameters)
result = self._substitute_params(result, parameters)
return result
def redact_sensitive(self, parameters: dict[str, Any], sensitive_keys: set[str]) -> dict[str, Any]:
"""Return a copy of parameters with sensitive values replaced by [REDACTED]."""
return {
k: "[REDACTED]" if k in sensitive_keys else v
for k, v in parameters.items()
}
# ── Conditional processing ────────────────────────────────────────────
def _process_conditionals(self, body: str, parameters: dict[str, Any]) -> str:
"""Process {% if param %} ... {% endif %} blocks."""
pattern = re.compile(
r'\{%\s*if\s+(\w+)\s*%\}(.*?)\{%\s*endif\s*%\}',
re.DOTALL
)
def replace_block(match: re.Match) -> str:
key = match.group(1)
content = match.group(2)
value = parameters.get(key)
# Truthy: non-empty string, non-empty list, True, non-zero number
if value is None or value == "" or value == [] or value is False:
return ""
return content
return pattern.sub(replace_block, body)
# ── Parameter substitution ────────────────────────────────────────────
# Matches {{ param }} or {{ param | filter }}
_PLACEHOLDER = re.compile(r'\{\{\s*(\w+)(?:\s*\|\s*(\w+))?\s*\}\}')
def _substitute_params(self, body: str, parameters: dict[str, Any]) -> str:
missing: list[str] = []
def replace(match: re.Match) -> str:
key = match.group(1)
filter_name = match.group(2)
if key not in parameters:
missing.append(key)
return match.group(0) # leave as-is, report at end
value = parameters[key]
if filter_name:
return self._apply_filter(filter_name, value)
return self._escape_string(value)
result = self._PLACEHOLDER.sub(replace, body)
if missing:
raise ScriptRenderError(
f"Missing required template parameter(s): {', '.join(missing)}"
)
return result
# ── Filters ──────────────────────────────────────────────────────────
def _apply_filter(self, filter_name: str, value: Any) -> str:
filters = {
"as_secure_string": self._filter_as_secure_string,
"as_array": self._filter_as_array,
"as_bool": self._filter_as_bool,
"escape_single": self._filter_escape_single,
}
if filter_name not in filters:
raise ScriptRenderError(f"Unknown template filter: {filter_name}")
return filters[filter_name](value)
def _filter_as_secure_string(self, value: Any) -> str:
escaped = str(value).replace("'", "''")
return f"(ConvertTo-SecureString '{escaped}' -AsPlainText -Force)"
def _filter_as_array(self, value: Any) -> str:
if not isinstance(value, list):
value = [value]
items = ",".join(f"'{self._escape_single_quote(str(v))}'" for v in value)
return items
def _filter_as_bool(self, value: Any) -> str:
if isinstance(value, bool):
return "$true" if value else "$false"
return "$true" if str(value).lower() in ("true", "1", "yes") else "$false"
def _filter_escape_single(self, value: Any) -> str:
return self._escape_single_quote(str(value))
# ── Escaping helpers ──────────────────────────────────────────────────
def _escape_string(self, value: Any) -> str:
"""Escape a value for safe insertion into a PowerShell single-quoted context."""
if isinstance(value, bool):
return "$true" if value else "$false"
if isinstance(value, (int, float)):
return str(value)
if isinstance(value, list):
# Lists rendered without filter — join with space (caller should use | as_array for proper PS arrays)
return " ".join(str(v) for v in value)
s = str(value)
# Escape backtick (PS escape char) first, then dollar (variable interpolation)
s = s.replace("`", "``")
s = s.replace("$", "`$")
# Escape single quotes (doubling for PS single-quoted strings)
s = self._escape_single_quote(s)
return s
def _escape_single_quote(self, s: str) -> str:
return s.replace("'", "''")

View File

@@ -20,7 +20,13 @@ from app.core.config import settings
settings.REQUIRE_INVITE_CODE = False
# Test database URL (separate from production)
TEST_DATABASE_URL = "postgresql+asyncpg://postgres:postgres@localhost:5432/patherly_test"
# Use DATABASE_TEST_URL env var if set (e.g. inside Docker where host is 'db'),
# otherwise fall back to localhost for local development.
import os
TEST_DATABASE_URL = os.environ.get(
"DATABASE_TEST_URL",
"postgresql+asyncpg://postgres:postgres@localhost:5432/patherly_test",
)
@pytest.fixture(scope="session")

View File

@@ -0,0 +1,114 @@
"""Tests for ScriptTemplateEngine — parameter substitution, sanitization, and filters."""
import pytest
from app.services.script_template_engine import ScriptTemplateEngine, ScriptRenderError
@pytest.fixture
def engine():
return ScriptTemplateEngine()
# ── Basic substitution ────────────────────────────────────────────────────
def test_simple_substitution(engine):
body = "New-ADUser -Name '{{ first_name }} {{ last_name }}'"
result = engine.render(body, {"first_name": "John", "last_name": "Smith"})
assert result == "New-ADUser -Name 'John Smith'"
def test_missing_required_param_raises(engine):
body = "New-ADUser -Name '{{ first_name }}'"
with pytest.raises(ScriptRenderError, match="first_name"):
engine.render(body, {})
def test_extra_params_ignored(engine):
body = "New-ADUser -Name '{{ first_name }}'"
result = engine.render(body, {"first_name": "John", "extra": "ignored"})
assert result == "New-ADUser -Name 'John'"
# ── Security: single-quote injection ─────────────────────────────────────
def test_single_quote_in_value_is_escaped(engine):
body = "Set-ADUser -Name '{{ name }}'"
result = engine.render(body, {"name": "O'Brien"})
# Single quotes doubled for PowerShell safety
assert "O''Brien" in result
def test_backtick_in_value_is_escaped(engine):
body = "Write-Host '{{ msg }}'"
result = engine.render(body, {"msg": "hello`world"})
assert "`" not in result or "``" in result # backtick is escaped
def test_dollar_sign_in_value_is_escaped(engine):
body = "Write-Host '{{ msg }}'"
result = engine.render(body, {"msg": "price is $100"})
# Dollar sign escaped so it doesn't interpolate as a PowerShell variable
assert "`$100" in result or "'price is $100'" in result
# ── Filters ──────────────────────────────────────────────────────────────
def test_as_secure_string_filter(engine):
body = "$secPwd = {{ password | as_secure_string }}"
result = engine.render(body, {"password": "MyP@ss123"})
assert "ConvertTo-SecureString" in result
assert "MyP@ss123" in result
assert "-AsPlainText -Force" in result
def test_as_array_filter(engine):
body = "$groups = @({{ groups | as_array }})"
result = engine.render(body, {"groups": ["GroupA", "GroupB"]})
assert "'GroupA','GroupB'" in result
def test_as_array_filter_single_item(engine):
body = "$groups = @({{ groups | as_array }})"
result = engine.render(body, {"groups": ["OnlyGroup"]})
assert "'OnlyGroup'" in result
def test_as_bool_filter_true(engine):
body = "$force = {{ force_change | as_bool }}"
result = engine.render(body, {"force_change": True})
assert "$true" in result
def test_as_bool_filter_false(engine):
body = "$force = {{ force_change | as_bool }}"
result = engine.render(body, {"force_change": False})
assert "$false" in result
# ── Conditional blocks ───────────────────────────────────────────────────
def test_if_block_included_when_truthy(engine):
body = "{% if groups %}\nAdd-Groups\n{% endif %}"
result = engine.render(body, {"groups": ["GroupA"]})
assert "Add-Groups" in result
def test_if_block_excluded_when_falsy(engine):
body = "{% if groups %}\nAdd-Groups\n{% endif %}"
result = engine.render(body, {"groups": []})
assert "Add-Groups" not in result
def test_if_block_excluded_when_missing(engine):
body = "{% if groups %}\nAdd-Groups\n{% endif %}"
result = engine.render(body, {})
assert "Add-Groups" not in result
# ── Parameter redaction ──────────────────────────────────────────────────
def test_sensitive_params_redacted_in_record(engine):
params = {"first_name": "John", "password": "Secret123"}
sensitive_keys = {"password"}
redacted = engine.redact_sensitive(params, sensitive_keys)
assert redacted["first_name"] == "John"
assert redacted["password"] == "[REDACTED]"

View File

@@ -0,0 +1,238 @@
"""Integration tests for Script Template Editor permissions and share endpoint."""
import pytest
from httpx import AsyncClient
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.models.user import User
from app.models.script_template import ScriptCategory, ScriptTemplate
# ── Helpers ──────────────────────────────────────────────────────────────
async def _create_category(db: AsyncSession) -> ScriptCategory:
"""Seed a script category for tests."""
cat = ScriptCategory(name="Active Directory", slug="active-directory", sort_order=1)
db.add(cat)
await db.commit()
await db.refresh(cat)
return cat
async def _make_owner(db: AsyncSession, user_id: str) -> None:
"""Promote a user to account owner."""
from uuid import UUID as PyUUID
result = await db.execute(select(User).where(User.id == PyUUID(user_id)))
user = result.scalar_one()
user.account_role = "owner"
await db.commit()
async def _register_and_login(client: AsyncClient, email: str, password: str, name: str) -> tuple[dict, str]:
"""Register a user, login, return (user_data, access_token)."""
resp = await client.post("/api/v1/auth/register", json={"email": email, "password": password, "name": name})
assert resp.status_code in (200, 201)
user_data = resp.json()
login_resp = await client.post("/api/v1/auth/login/json", json={"email": email, "password": password})
assert login_resp.status_code == 200
token = login_resp.json()["access_token"]
return user_data, token
TEMPLATE_PAYLOAD = {
"name": "Test Template",
"script_body": "Write-Host '{{ message }}'",
"parameters_schema": {
"parameters": [
{"key": "message", "label": "Message", "type": "text", "required": True, "order": 1}
]
},
"complexity": "beginner",
}
# ── Tests ────────────────────────────────────────────────────────────────
class TestScriptTemplatePermissions:
"""Test that engineers can create/edit their own templates, but not others'."""
@pytest.mark.asyncio
async def test_engineer_can_create_template(self, client, auth_headers, test_db):
cat = await _create_category(test_db)
payload = {**TEMPLATE_PAYLOAD, "category_id": str(cat.id)}
resp = await client.post("/api/v1/scripts/templates", json=payload, headers=auth_headers)
assert resp.status_code == 201
data = resp.json()
assert data["name"] == "Test Template"
assert data["created_by"] is not None
@pytest.mark.asyncio
async def test_engineer_can_edit_own_template(self, client, auth_headers, test_db):
cat = await _create_category(test_db)
payload = {**TEMPLATE_PAYLOAD, "category_id": str(cat.id)}
create_resp = await client.post("/api/v1/scripts/templates", json=payload, headers=auth_headers)
template_id = create_resp.json()["id"]
update_resp = await client.put(
f"/api/v1/scripts/templates/{template_id}",
json={"name": "Updated Template"},
headers=auth_headers,
)
assert update_resp.status_code == 200
assert update_resp.json()["name"] == "Updated Template"
@pytest.mark.asyncio
async def test_engineer_cannot_edit_others_template(self, client, test_db):
cat = await _create_category(test_db)
# Engineer A creates a template
_, token_a = await _register_and_login(client, "engineer_a@example.com", "TestPass123!", "Engineer A")
headers_a = {"Authorization": f"Bearer {token_a}"}
payload = {**TEMPLATE_PAYLOAD, "category_id": str(cat.id)}
create_resp = await client.post("/api/v1/scripts/templates", json=payload, headers=headers_a)
template_id = create_resp.json()["id"]
# Engineer B tries to edit it
_, token_b = await _register_and_login(client, "engineer_b@example.com", "TestPass123!", "Engineer B")
headers_b = {"Authorization": f"Bearer {token_b}"}
update_resp = await client.put(
f"/api/v1/scripts/templates/{template_id}",
json={"name": "Hijacked!"},
headers=headers_b,
)
assert update_resp.status_code == 403
@pytest.mark.asyncio
async def test_engineer_can_delete_own_template(self, client, auth_headers, test_db):
cat = await _create_category(test_db)
payload = {**TEMPLATE_PAYLOAD, "category_id": str(cat.id)}
create_resp = await client.post("/api/v1/scripts/templates", json=payload, headers=auth_headers)
template_id = create_resp.json()["id"]
delete_resp = await client.delete(f"/api/v1/scripts/templates/{template_id}", headers=auth_headers)
assert delete_resp.status_code == 204
@pytest.mark.asyncio
async def test_viewer_cannot_create_template(self, client, test_db):
_, token = await _register_and_login(client, "viewer@example.com", "TestPass123!", "Viewer")
# Downgrade to viewer
result = await test_db.execute(select(User).where(User.email == "viewer@example.com"))
user = result.scalar_one()
user.role = "viewer"
user.account_role = "viewer"
await test_db.commit()
# Re-login to get new token with updated role
login_resp = await client.post("/api/v1/auth/login/json", json={"email": "viewer@example.com", "password": "TestPass123!"})
token = login_resp.json()["access_token"]
headers = {"Authorization": f"Bearer {token}"}
cat = await _create_category(test_db)
payload = {**TEMPLATE_PAYLOAD, "category_id": str(cat.id)}
resp = await client.post("/api/v1/scripts/templates", json=payload, headers=headers)
assert resp.status_code == 403
@pytest.mark.asyncio
async def test_admin_can_edit_others_template(self, client, test_db, admin_auth_headers):
cat = await _create_category(test_db)
# Create template as a regular engineer
_, token = await _register_and_login(client, "eng@example.com", "TestPass123!", "Eng")
headers = {"Authorization": f"Bearer {token}"}
payload = {**TEMPLATE_PAYLOAD, "category_id": str(cat.id)}
create_resp = await client.post("/api/v1/scripts/templates", json=payload, headers=headers)
template_id = create_resp.json()["id"]
# Admin edits it
update_resp = await client.put(
f"/api/v1/scripts/templates/{template_id}",
json={"name": "Admin Updated"},
headers=admin_auth_headers,
)
assert update_resp.status_code == 200
assert update_resp.json()["name"] == "Admin Updated"
@pytest.mark.asyncio
async def test_managed_filter_returns_own_templates(self, client, test_db):
cat = await _create_category(test_db)
# Engineer A creates a template
_, token_a = await _register_and_login(client, "eng_a2@example.com", "TestPass123!", "Eng A")
headers_a = {"Authorization": f"Bearer {token_a}"}
payload = {**TEMPLATE_PAYLOAD, "category_id": str(cat.id)}
await client.post("/api/v1/scripts/templates", json=payload, headers=headers_a)
# Engineer B should not see A's template in managed view
_, token_b = await _register_and_login(client, "eng_b2@example.com", "TestPass123!", "Eng B")
headers_b = {"Authorization": f"Bearer {token_b}"}
resp = await client.get("/api/v1/scripts/templates?managed=true", headers=headers_b)
assert resp.status_code == 200
assert len(resp.json()) == 0
class TestScriptTemplateShare:
"""Test the share/unshare endpoint."""
@pytest.mark.asyncio
async def test_owner_can_share_template(self, client, test_db):
cat = await _create_category(test_db)
# Registration auto-sets account_role="owner", so this user is already an owner
user_data, token = await _register_and_login(client, "eng_share@example.com", "TestPass123!", "Eng")
headers = {"Authorization": f"Bearer {token}"}
payload = {**TEMPLATE_PAYLOAD, "category_id": str(cat.id)}
create_resp = await client.post("/api/v1/scripts/templates", json=payload, headers=headers)
template_id = create_resp.json()["id"]
# Share — owner should be allowed
share_resp = await client.patch(
f"/api/v1/scripts/templates/{template_id}/share?shared=true",
headers=headers,
)
assert share_resp.status_code == 200
@pytest.mark.asyncio
async def test_engineer_cannot_share_template(self, client, test_db):
cat = await _create_category(test_db)
# Create a user and downgrade to engineer (registration sets owner by default)
user_data, token = await _register_and_login(client, "eng_noshare@example.com", "TestPass123!", "Eng NoShare")
result = await test_db.execute(select(User).where(User.email == "eng_noshare@example.com"))
user = result.scalar_one()
user.account_role = "engineer"
await test_db.commit()
# Re-login to get fresh token with updated role
login_resp = await client.post("/api/v1/auth/login/json", json={"email": "eng_noshare@example.com", "password": "TestPass123!"})
eng_token = login_resp.json()["access_token"]
eng_headers = {"Authorization": f"Bearer {eng_token}"}
payload = {**TEMPLATE_PAYLOAD, "category_id": str(cat.id)}
create_resp = await client.post("/api/v1/scripts/templates", json=payload, headers=eng_headers)
template_id = create_resp.json()["id"]
share_resp = await client.patch(
f"/api/v1/scripts/templates/{template_id}/share?shared=true",
headers=eng_headers,
)
assert share_resp.status_code == 403
@pytest.mark.asyncio
async def test_owner_can_unshare_template(self, client, test_db):
cat = await _create_category(test_db)
# Registration auto-sets account_role="owner"
user_data, token = await _register_and_login(client, "eng_unshare@example.com", "TestPass123!", "Eng")
headers = {"Authorization": f"Bearer {token}"}
payload = {**TEMPLATE_PAYLOAD, "category_id": str(cat.id)}
create_resp = await client.post("/api/v1/scripts/templates", json=payload, headers=headers)
template_id = create_resp.json()["id"]
# Share then unshare
await client.patch(f"/api/v1/scripts/templates/{template_id}/share?shared=true", headers=headers)
unshare_resp = await client.patch(
f"/api/v1/scripts/templates/{template_id}/share?shared=false",
headers=headers,
)
assert unshare_resp.status_code == 200
assert unshare_resp.json()["team_id"] is None

View File

@@ -0,0 +1,336 @@
"""Integration tests for Script Generator API endpoints."""
import json
import uuid
from datetime import datetime, timezone
import pytest
import sqlalchemy as sa
# ── Fixtures ──────────────────────────────────────────────────────────────
@pytest.fixture
async def seed_script_data(test_db):
"""Seed script categories and templates into the test database."""
now = datetime.now(timezone.utc)
cat_id = uuid.UUID("00000000-0000-0000-0000-000000000001")
# Insert category
await test_db.execute(
sa.text("""
INSERT INTO script_categories (id, name, slug, description, icon, sort_order, is_active, created_at, updated_at)
VALUES (:id, :name, :slug, :description, :icon, :sort_order, true, :now, :now)
"""),
{
"id": cat_id,
"name": "Active Directory",
"slug": "active-directory",
"description": "User account and group management scripts",
"icon": "shield-check",
"sort_order": 1,
"now": now,
},
)
# Minimal template data for testing
templates = [
{
"id": uuid.UUID("00000000-0000-0000-0001-000000000001"),
"slug": "create-ad-user",
"name": "Create AD User Account",
"description": "Creates a new Active Directory user account.",
"script_body": "$SamAccountName = '{{ sam_account_name }}'",
"parameters_schema": json.dumps({
"parameters": [
{"key": "sam_account_name", "label": "SAM Account Name", "type": "text", "required": True, "order": 1},
]
}),
"complexity": "intermediate",
"estimated_runtime": "< 5 seconds",
"requires_elevation": True,
"tags": json.dumps(["active-directory", "user-management"]),
},
{
"id": uuid.UUID("00000000-0000-0000-0001-000000000002"),
"slug": "disable-ad-user",
"name": "Disable AD User Account",
"description": "Disables an Active Directory user account.",
"script_body": "$SamAccountName = '{{ sam_account_name }}'",
"parameters_schema": json.dumps({
"parameters": [
{"key": "sam_account_name", "label": "SAM Account Name", "type": "text", "required": True, "order": 1},
]
}),
"complexity": "beginner",
"estimated_runtime": "< 5 seconds",
"requires_elevation": True,
"tags": json.dumps(["active-directory", "offboarding"]),
},
{
"id": uuid.UUID("00000000-0000-0000-0001-000000000003"),
"slug": "reset-ad-password",
"name": "Reset AD Password",
"description": "Resets an Active Directory user password.",
"script_body": "$SamAccountName = '{{ sam_account_name }}'\n$NewPassword = {{ new_password | as_secure_string }}\n$ForceChange = {{ force_change_at_logon | as_bool }}\n$UnlockAccount = {{ unlock_account | as_bool }}",
"parameters_schema": json.dumps({
"parameters": [
{"key": "sam_account_name", "label": "SAM Account Name", "type": "text", "required": True, "order": 1},
{"key": "new_password", "label": "New Password", "type": "password", "required": True, "order": 2, "sensitive": True},
{"key": "force_change_at_logon", "label": "Force Change at Next Logon", "type": "boolean", "required": True, "order": 3},
{"key": "unlock_account", "label": "Unlock Account if Locked", "type": "boolean", "required": True, "order": 4},
]
}),
"complexity": "beginner",
"estimated_runtime": "< 5 seconds",
"requires_elevation": True,
"tags": json.dumps(["active-directory", "password"]),
},
{
"id": uuid.UUID("00000000-0000-0000-0001-000000000004"),
"slug": "unlock-ad-account",
"name": "Unlock AD Account",
"description": "Unlocks a locked-out Active Directory user account.",
"script_body": "$SamAccountName = '{{ sam_account_name }}'\n$ShowLockoutInfo = {{ show_lockout_info | as_bool }}",
"parameters_schema": json.dumps({
"parameters": [
{"key": "sam_account_name", "label": "SAM Account Name", "type": "text", "required": True, "order": 1},
{"key": "show_lockout_info", "label": "Show Lockout Source Info", "type": "boolean", "required": False, "order": 2},
]
}),
"complexity": "beginner",
"estimated_runtime": "< 5 seconds",
"requires_elevation": True,
"tags": json.dumps(["active-directory", "lockout"]),
},
{
"id": uuid.UUID("00000000-0000-0000-0001-000000000005"),
"slug": "delete-ad-user",
"name": "Delete AD User Account",
"description": "Permanently deletes an Active Directory user account.",
"script_body": "$SamAccountName = '{{ sam_account_name }}'\n$ConfirmDeletion = {{ confirm_deletion | as_bool }}",
"parameters_schema": json.dumps({
"parameters": [
{"key": "sam_account_name", "label": "SAM Account Name", "type": "text", "required": True, "order": 1},
{"key": "confirm_deletion", "label": "Confirm Deletion", "type": "boolean", "required": True, "order": 2},
]
}),
"complexity": "advanced",
"estimated_runtime": "< 10 seconds",
"requires_elevation": True,
"tags": json.dumps(["active-directory", "destructive"]),
},
{
"id": uuid.UUID("00000000-0000-0000-0001-000000000006"),
"slug": "bulk-user-import",
"name": "Bulk User Import from CSV",
"description": "Imports multiple Active Directory user accounts from a CSV file.",
"script_body": "$CSVPath = '{{ csv_path }}'\n$OUPath = '{{ ou_path }}'",
"parameters_schema": json.dumps({
"parameters": [
{"key": "csv_path", "label": "CSV File Path", "type": "text", "required": True, "order": 1},
{"key": "ou_path", "label": "Target OU", "type": "text", "required": True, "order": 2},
]
}),
"complexity": "advanced",
"estimated_runtime": "1-2 minutes",
"requires_elevation": True,
"tags": json.dumps(["active-directory", "bulk"]),
},
]
for tmpl in templates:
await test_db.execute(
sa.text("""
INSERT INTO script_templates (
id, category_id, name, slug, description,
script_body, parameters_schema, default_values, validation_rules,
tags, complexity, estimated_runtime, requires_elevation,
requires_modules, version, is_verified, is_active, usage_count,
created_at, updated_at
) VALUES (
:id, :category_id, :name, :slug, :description,
:script_body, CAST(:parameters_schema AS jsonb), '{}'::jsonb, '{}'::jsonb,
CAST(:tags AS jsonb), :complexity, :estimated_runtime, :requires_elevation,
'[]'::jsonb, 1, true, true, 0,
:now, :now
)
"""),
{**tmpl, "category_id": cat_id, "now": now},
)
await test_db.commit()
return cat_id
# ── Categories ────────────────────────────────────────────────────────────
@pytest.mark.asyncio
async def test_list_categories_requires_auth(client):
response = await client.get("/api/v1/scripts/categories")
assert response.status_code == 401
@pytest.mark.asyncio
async def test_list_categories_returns_seeded_data(client, auth_headers, seed_script_data):
response = await client.get("/api/v1/scripts/categories", headers=auth_headers)
assert response.status_code == 200
data = response.json()
assert isinstance(data, list)
assert any(c["slug"] == "active-directory" for c in data)
# ── Templates ─────────────────────────────────────────────────────────────
@pytest.mark.asyncio
async def test_list_templates_requires_auth(client):
response = await client.get("/api/v1/scripts/templates")
assert response.status_code == 401
@pytest.mark.asyncio
async def test_list_templates_returns_seeded_data(client, auth_headers, seed_script_data):
response = await client.get("/api/v1/scripts/templates", headers=auth_headers)
assert response.status_code == 200
data = response.json()
assert len(data) == 6
slugs = [t["slug"] for t in data]
assert "create-ad-user" in slugs
assert "reset-ad-password" in slugs
@pytest.mark.asyncio
async def test_list_templates_filter_by_category(client, auth_headers, seed_script_data):
response = await client.get(
"/api/v1/scripts/templates?category_slug=active-directory",
headers=auth_headers,
)
assert response.status_code == 200
data = response.json()
assert len(data) == 6
@pytest.mark.asyncio
async def test_list_templates_search(client, auth_headers, seed_script_data):
response = await client.get(
"/api/v1/scripts/templates?search=password",
headers=auth_headers,
)
assert response.status_code == 200
data = response.json()
assert any("password" in t["name"].lower() or "password" in t["slug"] for t in data)
@pytest.mark.asyncio
async def test_get_template_detail(client, auth_headers, seed_script_data):
list_resp = await client.get("/api/v1/scripts/templates", headers=auth_headers)
templates = list_resp.json()
template_id = templates[0]["id"]
response = await client.get(f"/api/v1/scripts/templates/{template_id}", headers=auth_headers)
assert response.status_code == 200
data = response.json()
assert "script_body" in data
assert "parameters_schema" in data
@pytest.mark.asyncio
async def test_get_template_detail_not_found(client, auth_headers):
response = await client.get(
"/api/v1/scripts/templates/00000000-0000-0000-0000-000000000099",
headers=auth_headers,
)
assert response.status_code == 404
# ── Generate ──────────────────────────────────────────────────────────────
@pytest.mark.asyncio
async def test_generate_script_success(client, auth_headers, seed_script_data):
list_resp = await client.get(
"/api/v1/scripts/templates?search=unlock",
headers=auth_headers,
)
unlock_template = list_resp.json()[0]
response = await client.post(
"/api/v1/scripts/generate",
json={
"template_id": unlock_template["id"],
"parameters": {"sam_account_name": "jsmith", "show_lockout_info": False},
},
headers=auth_headers,
)
assert response.status_code == 200
data = response.json()
assert "script" in data
assert "jsmith" in data["script"]
assert "id" in data
@pytest.mark.asyncio
async def test_generate_script_missing_required_param(client, auth_headers, seed_script_data):
list_resp = await client.get(
"/api/v1/scripts/templates?search=unlock",
headers=auth_headers,
)
unlock_template = list_resp.json()[0]
response = await client.post(
"/api/v1/scripts/generate",
json={
"template_id": unlock_template["id"],
"parameters": {},
},
headers=auth_headers,
)
assert response.status_code == 422
@pytest.mark.asyncio
async def test_generate_script_password_redacted_in_record(client, auth_headers, seed_script_data):
list_resp = await client.get(
"/api/v1/scripts/templates?search=reset-ad-password",
headers=auth_headers,
)
reset_template = list_resp.json()[0]
await client.post(
"/api/v1/scripts/generate",
json={
"template_id": reset_template["id"],
"parameters": {
"sam_account_name": "jsmith",
"new_password": "SuperSecret123!",
"force_change_at_logon": True,
"unlock_account": True,
},
},
headers=auth_headers,
)
history_resp = await client.get("/api/v1/scripts/generations", headers=auth_headers)
assert history_resp.status_code == 200
generations = history_resp.json()
assert len(generations) > 0
latest = generations[0]
assert latest["parameters_used"].get("new_password") == "[REDACTED]"
# ── Team template CRUD ────────────────────────────────────────────────────
@pytest.mark.asyncio
async def test_create_team_template_requires_team_admin(client, auth_headers, seed_script_data):
list_resp = await client.get("/api/v1/scripts/categories", headers=auth_headers)
cat_id = list_resp.json()[0]["id"]
response = await client.post(
"/api/v1/scripts/templates",
json={
"category_id": cat_id,
"name": "My Custom Script",
"script_body": "Write-Host 'hello'",
"parameters_schema": {},
},
headers=auth_headers, # regular engineer
)
assert response.status_code == 403