Runbook Automation Bots

import os
from slack_bolt import App
from slack_bolt.adapter.fastapi import SlackRequestHandler

 # Initialize the Slack app
app = App(
    token=os.environ["SLACK_BOT_TOKEN"],
    signing_secret=os.environ["SLACK_SIGNING_SECRET"],
)

@app.command("/ops")
def handle_ops_command(ack, command, say):
    """Handle /ops slash command."""
    ack()
    text = command["text"]
    parts = text.split()
    action = parts[0] if parts else "help"

    if action == "health":
        say("Checking service health...")
        result = run_health_check()
        say(f"Health check results:\n{result}")
    elif action == "restart":
        if len(parts) < 2:
            say("Usage: /ops restart <service-name>")
            return
        service = parts[1]
        say(f"Restarting {service}...")
        result = restart_service(service)
        say(f"Restart result: {result}")
    else:
        say("Available commands: health, restart <service>, scale, logs")

 # Run with FastAPI
from fastapi import FastAPI, Request

fastapi_app = FastAPI()
handler = SlackRequestHandler(app)

@fastapi_app.post("/slack/events")
async def slack_events(request: Request):
    return await handler.handle(request)

import subprocess
import httpx
from kubernetes import client, config

@app.command("/ops")
def ops_handler(ack, command, say, respond):
    ack()
    parts = command["text"].split()
    action = parts[0] if parts else "help"

    commands = {
        "health": cmd_health,
        "restart": cmd_restart,
        "scale": cmd_scale,
        "logs": cmd_logs,
        "deploy": cmd_deploy,
        "rollback": cmd_rollback,
        "silence": cmd_silence,
        "status": cmd_status,
    }

    handler = commands.get(action)
    if handler:
        try:
            result = handler(parts[1:], command)
            say(result)
        except Exception as e:
            say(f"Error: {e}")
    else:
        say(f"Unknown command: {action}. Available: {', '.join(commands.keys())}")

def cmd_health(args, command) -> str:
    """Check service health."""
    service = args[0] if args else "all"
    if service == "all":
         # Check all services
        results = []
        for svc in ["api", "web", "worker"]:
            status = check_service(svc)
            results.append(f"{svc}: {status}")
        return "\n".join(results)
    return f"{service}: {check_service(service)}"

def cmd_restart(args, command) -> str:
    """Restart a service."""
    if not args:
        return "Usage: /ops restart <service>"
    service = args[0]
    user = command["user_id"]
    log_action(f"User {user} restarting {service}")
    result = restart_service(service)
    return f"Restarted {service}: {result}"

from slack_sdk import WebClient
from datetime import datetime

slack = WebClient(token=os.environ["SLACK_BOT_TOKEN"])

@app.command("/incident")
def incident_handler(ack, command, say):
    ack()
    parts = command["text"].split()
    severity = parts[0] if parts else "unknown"
    title = " ".join(parts[1:]) if len(parts) > 1 else "No title"

     # Create incident channel
    channel = slack.conversations_create(
        name=f"inc-{datetime.now().strftime('%Y%m%d')}-{title[:20].replace(' ', '-').lower()}",
    )
    channel_id = channel["channel"]["id"]

     # Post initial message
    slack.chat_postMessage(
        channel=channel_id,
        text=f"*Incident: {title}*\nSeverity: {severity}\nCreated by: <@{command['user_id']}>",
        blocks=[
            {"type": "header", "text": {"type": "plain_text", "text": f"Incident: {title}"}},
            {"type": "section", "fields": [
                {"type": "mrkdwn", "text": f"*Severity:* {severity}"},
                {"type": "mrkdwn", "text": f"*Created:* {datetime.now().isoformat()}"},
                {"type": "mrkdwn", "text": f"*Created by:* <@{command['user_id']}>"},
            ]},
            {"type": "section", "text": {"type": "mrkdwn", "text": "Commands:\n• `/ops health` — Check service health\n• `/ops restart <service>` — Restart a service\n• `/ops logs <service>` — View recent logs"}},
        ],
    )

    say(f"Incident channel created: <#{channel_id}>")

from slack_sdk import WebClient

slack = WebClient(token=os.environ["SLACK_BOT_TOKEN"])

@app.command("/ops/auto-fix")
def auto_fix_handler(ack, command, say):
    ack()
    parts = command["text"].split()
    if not parts:
        say("Usage: /ops auto-fix <alert-name>")
        return

    alert_name = parts[0]

     # Check if auto-fix is available for this alert
    remediation = REMEDIATION_MAP.get(alert_name)
    if not remediation:
        say(f"No auto-fix available for {alert_name}. Follow the runbook manually.")
        return

     # Send confirmation request
    slack.chat_postMessage(
        channel=command["channel_id"],
        blocks=[
            {"type": "section", "text": {"type": "mrkdwn", "text": f"*Auto-fix available for {alert_name}:*\n{remediation['description']}"}},
            {"type": "actions", "elements": [
                {"type": "button", "text": {"type": "plain_text", "text": "Confirm"}, "style": "primary", "action_id": f"confirm_fix_{alert_name}"},
                {"type": "button", "text": {"type": "plain_text", "text": "Cancel"}, "style": "danger", "action_id": f"cancel_fix_{alert_name}"},
            ]},
        ],
    )

@app.action("confirm_fix_.*")
def handle_fix_confirmation(ack, action, say):
    ack()
    alert_name = action["action_id"].replace("confirm_fix_", "")
    remediation = REMEDIATION_MAP[alert_name]

    say(f"Executing remediation for {alert_name}...")
    try:
        result = remediation["function"]()
        say(f"Remediation complete: {result}")
    except Exception as e:
        say(f"Remediation failed: {e}")

REMEDIATION_MAP = {
    "HighMemory": {"description": "Restart the service to clear memory", "function": lambda: restart_service("api")},
    "DiskFull": {"description": "Clean old logs from /var/log", "function": lambda: clean_logs()},
    "PodCrashLoop": {"description": "Restart the pod", "function": lambda: restart_pod("api-pod")},
}

import os
from functools import wraps

 # Only allow certain users to run dangerous commands
ALLOWED_USERS = os.environ.get("OPS_ALLOWED_USERS", "").split(",")
DANGEROUS_COMMANDS = {"restart", "scale", "deploy", "rollback"}

def require_permission(func):
    """Check if user is allowed to run the command."""
    @wraps(func)
    def wrapper(args, command):
        user = command["user_id"]
        action = args[0] if args else ""

        if action in DANGEROUS_COMMANDS and user not in ALLOWED_USERS:
            return f"Permission denied. Only {', '.join(ALLOWED_USERS)} can run {action}."

        return func(args, command)
    return wrapper

 # Rate limiting
import time
from collections import defaultdict

command_timestamps = defaultdict(list)

def rate_limit(max_per_minute: int = 5):
    """Limit commands per user per minute."""
    def decorator(func):
        @wraps(func)
        def wrapper(args, command):
            user = command["user_id"]
            now = time.time()
            timestamps = command_timestamps[user]

             # Remove timestamps older than 1 minute
            timestamps[:] = [t for t in timestamps if now - t < 60]

            if len(timestamps) >= max_per_minute:
                return f"Rate limit exceeded. Max {max_per_minute} commands per minute."

            timestamps.append(now)
            return func(args, command)
        return wrapper
    return decorator

 # Audit logging
def log_action(user: str, action: str, target: str, result: str):
    """Log all actions for audit trail."""
    import json
    from datetime import datetime

    entry = {
        "timestamp": datetime.utcnow().isoformat(),
        "user": user,
        "action": action,
        "target": target,
        "result": result,
    }
    with open("/var/log/ops-bot/audit.jsonl", "a") as f:
        f.write(json.dumps(entry) + "\n")