11.2.2 Automation Script
Definisi
Automation Script adalah fondasi operasi keamanan modern - menjadwalkan dan mengeksekusi tugas keamanan secara otomatis tanpa intervensi manual. Mulai dari cron job sederhana untuk backup harian, systemd timer untuk service monitoring, CI/CD pipeline untuk security testing, hingga Infrastructure as Code (IaC) dengan Ansible dan Terraform. Dalam dunia cybersecurity, automation script memungkinkan response insiden dalam hitungan detik, scanning terjadwal, dan enforce kebijakan keamanan secara konsisten.
1. Cron - Job Scheduler Linux
Dasar Cron
# Format cron: menit jam tanggal bulan hari command
# Contoh:
# 0 2 * * * /home/security/scripts/scan.sh
# ^ ^ ^ ^ ^
# | | | | +-- Hari (0=Minggu, 1=Senin, ... 6=Sabtu)
# | | | +---- Bulan (1-12)
# | | +------ Tanggal (1-31)
# | +-------- Jam (0-23)
# +---------- Menit (0-59)
Special Strings
@reboot # Jalankan saat boot
@hourly # Setiap jam (0 * * * *)
@daily # Setiap hari jam 00:00 (0 0 * * *)
@weekly # Setiap Minggu jam 00:00 (0 0 * * 0)
@monthly # Setiap tanggal 1 jam 00:00 (0 0 1 * *)
@yearly # Setiap 1 Januari jam 00:00 (0 0 1 1 *)
Contoh Cron untuk Security
# Setiap 30 menit - scan log brute force
*/30 * * * * /home/security/scripts/brute-force-detect.sh
# Setiap jam - update threat intelligence
0 * * * * /home/security/scripts/update-threat-intel.py
# Setiap hari jam 2 pagi - backup konfigurasi
0 2 * * * /home/security/scripts/auto-backup.sh
# Setiap hari jam 8 pagi - kirim daily report
0 8 * * * /home/security/scripts/daily-report.py
# Setiap Senin jam 9 pagi - full vulnerability scan
0 9 * * 1 /home/security/scripts/nmap-full-scan.sh
# Setiap tanggal 1 jam 3 pagi - log rotation manual
0 3 1 * * /home/security/scripts/manual-logrotate.sh
Manajemen Cron
# Edit crontab user saat ini
crontab -e
# List semua cron jobs
crontab -l
# Hapus semua cron jobs
crontab -r
# Edit crontab untuk user lain (root)
sudo crontab -e -u www-data
# Backup cron jobs
crontab -l > ~/cron-backup-$(date +%Y%m%d).txt
# Restore cron jobs
crontab ~/cron-backup-20260722.txt
Script Cron yang Baik
#!/bin/bash
# safe-cron-script.sh
# Template script yang aman untuk cron job
set -euo pipefail
IFS=$'\n\t'
# Konfigurasi
LOG_FILE="/var/log/security-scripts/$(basename $0).log"
PID_FILE="/tmp/$(basename $0).pid"
# Cek lock file - cegah multiple instance
if [ -f "$PID_FILE" ]; then
echo "[$(date)] Script masih berjalan, skip" >> "$LOG_FILE"
exit 1
fi
# Buat lock file
echo $$ > "$PID_FILE"
trap 'rm -f "$PID_FILE"; exit' EXIT INT TERM
# Logging
exec 1>> "$LOG_FILE" 2>&1
echo "[$(date)] Mulai eksekusi"
# Main logic
# ... script di sini ...
echo "[$(date)] Selesai"
2. Systemd Timer - Alternatif Modern Cron
Systemd timer menawarkan fleksibilitas lebih dari cron: dependency management, logging terintegrasi, dan monitoring service.
Service Unit - auto-scan.service
[Unit]
Description=Auto Security Scan Service
Documentation=https://docs.security.local
[Service]
Type=oneshot
ExecStart=/usr/local/bin/auto-scan.sh
User=security
Group=security
StandardOutput=journal
StandardError=journal
# Security hardening
ProtectSystem=strict
ReadWritePaths=/var/log/scans
PrivateTmp=true
NoNewPrivileges=true
[Install]
WantedBy=multi-user.target
Timer Unit - auto-scan.timer
[Unit]
Description=Run auto-scan setiap 6 jam
Requires=auto-scan.service
[Timer]
# Setiap 6 jam
OnCalendar=*-*-* 00,06,12,18:00:00
# Atau: setiap 30 menit sejak boot
# OnBootSec=5min
# OnUnitActiveSec=30min
# Random delay untuk cegah thundering herd
RandomizedDelaySec=5min
# Persistent - kejar jadwal yang terlewat
Persistent=true
[Install]
WantedBy=timers.target
Manajemen Systemd Timer
# Reload daemon
sudo systemctl daemon-reload
# Enable dan start timer
sudo systemctl enable auto-scan.timer
sudo systemctl start auto-scan.timer
# Cek status timer
sudo systemctl status auto-scan.timer
# List semua timer
systemctl list-timers --all
# Lihat log service
journalctl -u auto-scan.service
# Lihat log real-time
journalctl -u auto-scan.service -f
# Trigger manual
sudo systemctl start auto-scan.service
# Stop timer
sudo systemctl stop auto-scan.timer
sudo systemctl disable auto-scan.timer
Contoh Lengkap: Auto Backup dengan Systemd
#!/bin/bash
# /usr/local/bin/auto-backup.sh
# Auto backup dengan systemd timer
BACKUP_DIR="/backup/security"
DATE=$(date +%Y%m%d_%H%M%S)
RETENTION_DAYS=30
# Buat direktori backup
mkdir -p "$BACKUP_DIR"
# Backup konfigurasi penting
tar -czf "$BACKUP_DIR/configs_$DATE.tar.gz" \
/etc/nginx \
/etc/ssh \
/etc/fail2ban \
/etc/iptables
# Backup database SIEM
pg_dump siem_db > "$BACKUP_DIR/siem_db_$DATE.sql"
gzip "$BACKUP_DIR/siem_db_$DATE.sql"
# Hapus backup lebih dari 30 hari
find "$BACKUP_DIR" -name "*.tar.gz" -mtime +$RETENTION_DAYS -delete
find "$BACKUP_DIR" -name "*.sql.gz" -mtime +$RETENTION_DAYS -delete
echo "[OK] Backup selesai: $BACKUP_DIR/configs_$DATE.tar.gz"
3. inotify - File System Watcher
inotify memonitor perubahan filesystem secara real-time - ideal untuk deteksi file mencurigakan, perubahan konfigurasi, atau response otomatis.
inotifywait - CLI Watch
# Install
sudo apt install inotify-tools
# Monitor perubahan file
inotifywait -m /etc/ssh -e modify,create,delete |
while read directory event file; do
echo "[ALERT] $event di $directory$file"
echo " Waktu: $(date)"
echo " Cek integritas konfigurasi SSH!"
done
# Monitor directory log
inotifywait -m /var/log -e modify |
while read dir ev file; do
if [[ "$file" == "auth.log" ]]; then
tail -1 /var/log/auth.log | grep "Failed password"
fi
done
Script: Real-Time File Integrity Monitor
#!/bin/bash
# file-integrity-monitor.sh
# Monitor perubahan file konfigurasi penting
WATCH_DIRS="/etc/ssh /etc/nginx /etc/fail2ban /etc/iptables"
ALERT_SCRIPT="/usr/local/bin/send-alert.sh"
HASH_DB="/var/lib/file-monitor/hashes.db"
# Buat baseline hash jika belum ada
mkdir -p "$(dirname "$HASH_DB")"
if [ ! -f "$HASH_DB" ]; then
echo "[INIT] Membuat baseline hash..."
for dir in $WATCH_DIRS; do
find "$dir" -type f -exec sha256sum {} \; >> "$HASH_DB"
done
echo "[INIT] Baseline selesai - $(wc -l < "$HASH_DB") files"
exit 0
fi
# Monitor perubahan
inotifywait -m -r $WATCH_DIRS -e modify,create,delete,move |
while read path event file; do
fullpath="$path$file"
echo "[$(date)] $event: $fullpath"
# Hitung hash baru dan bandingkan
if [ -f "$fullpath" ]; then
new_hash=$(sha256sum "$fullpath" | awk '{print $1}')
old_hash=$(grep "$fullpath" "$HASH_DB" | awk '{print $1}')
if [ "$new_hash" != "$old_hash" ]; then
echo "[ALERT] File berubah: $fullpath"
# Kirim notifikasi
$ALERT_SCRIPT "File Integrity" "File berubah: $fullpath"
# Update hash DB
sed -i "/$fullpath/d" "$HASH_DB"
echo "$new_hash $fullpath" >> "$HASH_DB"
fi
fi
done
Python Watchdog - Cross-Platform Watcher
#!/usr/bin/env python3
# config-watcher.py
# Monitor konfigurasi dengan Python watchdog
import time
import hashlib
import logging
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
import requests
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(message)s"
)
logger = logging.getLogger(__name__)
class ConfigSecurityHandler(FileSystemEventHandler):
"""Handler untuk event filesystem dengan security context."""
def __init__(self, webhook_url=None):
self.webhook_url = webhook_url
self.hash_cache = {}
def get_file_hash(self, path):
"""Hitung SHA256 file."""
try:
with open(path, "rb") as f:
return hashlib.sha256(f.read()).hexdigest()
except (IOError, FileNotFoundError):
return None
def send_alert(self, event_type, path):
"""Kirim alert via webhook."""
if not self.webhook_url:
return
message = {
"text": (
f"[SECURITY ALERT] Config Change Detected!\n"
f"Event: {event_type}\n"
f"File: {path}\n"
f"Time: {time.strftime('%Y-%m-%d %H:%M:%S')}"
)
}
try:
requests.post(self.webhook_url, json=message, timeout=5)
except Exception as e:
logger.error(f"Gagal kirim alert: {e}")
def on_modified(self, event):
if event.is_directory:
return
new_hash = self.get_file_hash(event.src_path)
old_hash = self.hash_cache.get(event.src_path)
if new_hash and new_hash != old_hash:
logger.warning(f"MODIFIED: {event.src_path}")
logger.warning(f" Old hash: {old_hash}")
logger.warning(f" New hash: {new_hash}")
self.send_alert("MODIFIED", event.src_path)
self.hash_cache[event.src_path] = new_hash
def on_created(self, event):
if not event.is_directory:
logger.warning(f"CREATED: {event.src_path}")
self.hash_cache[event.src_path] = self.get_file_hash(event.src_path)
self.send_alert("CREATED", event.src_path)
def on_deleted(self, event):
if not event.is_directory:
logger.warning(f"DELETED: {event.src_path}")
self.hash_cache.pop(event.src_path, None)
self.send_alert("DELETED", event.src_path)
# Setup watcher
if __name__ == "__main__":
watch_paths = ["/etc/ssh", "/etc/nginx", "/etc/fail2ban"]
event_handler = ConfigSecurityHandler(
webhook_url="https://hooks.slack.com/services/T00/B00/xxxx"
)
observers = []
for path in watch_paths:
observer = Observer()
observer.schedule(event_handler, path, recursive=True)
observer.start()
observers.append(observer)
logger.info(f"Watching: {path}")
try:
while True:
time.sleep(1)
except KeyboardInterrupt:
for observer in observers:
observer.stop()
for observer in observers:
observer.join()
4. CI/CD Pipeline - GitHub Actions untuk Security
GitHub Actions memungkinkan automation security testing di setiap push, PR, dan release.
Workflow: Auto Security Scan
# .github/workflows/security-scan.yml
name: Auto Security Scan
on:
push:
branches: [main, develop]
pull_request:
branches: [main]
schedule:
# Scan full setiap hari Minggu jam 3 pagi
- cron: '0 3 * * 0'
jobs:
static-analysis:
name: Static Analysis
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- name: Bandit - Python Security Linter
run: |
pip install bandit
bandit -r . -f json -o bandit-report.json
- name: Semgrep - SAST Scan
uses: semgrep/semgrep-action@v1
with:
config: p/owasp-top-ten
- name: TruffleHog - Secret Detection
uses: trufflesecurity/trufflehog@v3
with:
extra_args: --results=verified,unknown
dependency-scan:
name: Dependency Scan
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- name: Trivy - Vulnerability Scanner
uses: aquasecurity/trivy-action@master
with:
scan-type: fs
scan-ref: .
format: sarif
output: trivy-results.sarif
- name: Snyk - Dependency Check
uses: snyk/actions/python@master
env:
SNYK_TOKEN: ${{ secrets.SNYK_TOKEN }}
with:
args: --severity-threshold=high
notify:
needs: [static-analysis, dependency-scan]
if: failure()
runs-on: ubuntu-latest
steps:
- name: Send Telegram Alert
uses: appleboy/telegram-action@master
with:
to: ${{ secrets.TELEGRAM_CHAT_ID }}
token: ${{ secrets.TELEGRAM_TOKEN }}
message: |
🔴 Security Scan FAILED!
Repo: ${{ github.repository }}
Branch: ${{ github.ref_name }}
Commit: ${{ github.sha }}
URL: ${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }}
Workflow: Auto Deploy dengan Security Gate
# .github/workflows/deploy-with-security.yml
name: Deploy with Security Gate
on:
push:
tags:
- 'v*'
jobs:
security-gate:
name: Security Gate
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- name: SAST Scan
run: |
docker run --rm -v $(pwd):/src semgrep \
semgrep --config=auto --error /src
- name: Container Scan
run: |
docker build -t app:test .
docker run --rm -v /var/run/docker.sock:/var/run/docker.sock \
aquasec/trivy image --severity CRITICAL,HIGH app:test
- name: Infrastructure as Code Scan
uses: bridgecrewio/checkov-action@master
with:
directory: terraform/
framework: terraform
deploy:
name: Deploy to Production
needs: security-gate
runs-on: ubuntu-latest
environment: production
steps:
- name: Deploy via SSH
uses: appleboy/ssh-action@master
with:
host: ${{ secrets.DEPLOY_HOST }}
username: ${{ secrets.DEPLOY_USER }}
key: ${{ secrets.DEPLOY_KEY }}
script: |
cd /opt/app
git pull
docker compose pull
docker compose up -d
echo "Deploy selesai: $(date)"
- name: Health Check
run: |
for i in {1..12}; do
if curl -sf https://app.company.com/health; then
echo "Health check PASSED"
exit 0
fi
sleep 5
done
echo "Health check FAILED"
exit 1
5. Ansible - Configuration Automation
Ansible mengotomatiskan konfigurasi dan enforce kebijakan keamanan di banyak server sekaligus.
Struktur Project Ansible
ansible-security/
├── ansible.cfg
├── inventory/
│ ├── production.yml
│ └── staging.yml
├── playbooks/
│ ├── hardening.yml
│ ├── firewall-rules.yml
│ └── log-audit.yml
└── roles/
├── ssh-hardening/
│ ├── tasks/
│ │ └── main.yml
│ └── templates/
│ └── sshd_config.j2
└── fail2ban/
├── tasks/
│ └── main.yml
└── templates/
└── jail.conf.j2
Playbook: SSH Hardening
# playbooks/ssh-hardening.yml
---
- name: SSH Hardening - Security Baseline
hosts: all
become: yes
vars:
ssh_port: 2222
allowed_users:
- admin
- security
banner_message: |
WARNING: Authorized access only.
All activities are monitored and logged.
tasks:
- name: Install OpenSSH server
apt:
name: openssh-server
state: present
update_cache: yes
- name: Deploy SSH config
template:
src: sshd_config.j2
dest: /etc/ssh/sshd_config
owner: root
group: root
mode: 0600
notify: restart sshd
- name: Set banner
copy:
content: "{{ banner_message }}"
dest: /etc/ssh/banner
owner: root
group: root
mode: 0644
- name: Ensure fail2ban running
systemd:
name: fail2ban
state: started
enabled: yes
handlers:
- name: restart sshd
systemd:
name: sshd
state: restarted
Playbook: Firewall Rules
# playbooks/firewall-rules.yml
---
- name: Apply Firewall Rules
hosts: all
become: yes
vars:
allowed_ports:
- { port: 22, proto: tcp, comment: "SSH" }
- { port: 80, proto: tcp, comment: "HTTP" }
- { port: 443, proto: tcp, comment: "HTTPS" }
- { port: 514, proto: tcp, comment: "Syslog" }
tasks:
- name: Flush existing rules
iptables:
chain: "{{ item }}"
flush: yes
loop:
- INPUT
- FORWARD
- OUTPUT
- name: Set default policies
iptables:
chain: "{{ item.chain }}"
policy: "{{ item.policy }}"
loop:
- { chain: INPUT, policy: DROP }
- { chain: FORWARD, policy: DROP }
- { chain: OUTPUT, policy: ACCEPT }
- name: Allow established connections
iptables:
chain: INPUT
ctstate: ESTABLISHED,RELATED
jump: ACCEPT
- name: Allow loopback
iptables:
chain: INPUT
in_interface: lo
jump: ACCEPT
- name: Allow configured ports
iptables:
chain: INPUT
protocol: "{{ item.proto }}"
destination_port: "{{ item.port }}"
jump: ACCEPT
comment: "{{ item.comment }}"
loop: "{{ allowed_ports }}"
- name: Save rules
shell: iptables-save > /etc/iptables/rules.v4
Run Ansible Playbook
# Test koneksi
ansible all -i inventory/production.yml -m ping
# Dry run
ansible-playbook -i inventory/production.yml \
playbooks/ssh-hardening.yml --check --diff
# Apply
ansible-playbook -i inventory/production.yml \
playbooks/ssh-hardening.yml
# Apply dengan limit ke grup tertentu
ansible-playbook -i inventory/production.yml \
playbooks/firewall-rules.yml --limit webservers
# Dengan verbose logging
ansible-playbook -i inventory/production.yml \
playbooks/log-audit.yml -vvv
6. Terraform - Infrastructure as Code
Terraform mengelola infrastruktur keamanan secara deklaratif: firewall rules, security groups, IAM policies.
Contoh: Security Group dengan Terraform
# main.tf
terraform {
required_providers {
aws = {
source = "hashicorp/aws"
version = "~> 5.0"
}
}
}
provider "aws" {
region = var.aws_region
}
# Security Group - Web Server
resource "aws_security_group" "web_server" {
name = "web-server-sg"
description = "Security group untuk web server"
vpc_id = aws_vpc.main.id
tags = {
Name = "web-server-sg"
Managed = "terraform"
}
}
# Ingress rules
resource "aws_vpc_security_group_ingress_rule" "allow_http" {
security_group_id = aws_security_group.web_server.id
cidr_ipv4 = "0.0.0.0/0"
from_port = 80
to_port = 80
ip_protocol = "tcp"
}
resource "aws_vpc_security_group_ingress_rule" "allow_https" {
security_group_id = aws_security_group.web_server.id
cidr_ipv4 = "0.0.0.0/0"
from_port = 443
to_port = 443
ip_protocol = "tcp"
}
resource "aws_vpc_security_group_ingress_rule" "allow_ssh" {
security_group_id = aws_security_group.web_server.id
cidr_ipv4 = var.admin_cidr
from_port = 22
to_port = 22
ip_protocol = "tcp"
}
# Egress - allow all outbound
resource "aws_vpc_security_group_egress_rule" "allow_all" {
security_group_id = aws_security_group.web_server.id
cidr_ipv4 = "0.0.0.0/0"
ip_protocol = "-1" # Semua protokol
}
# WAF - Web Application Firewall
resource "aws_wafv2_web_acl" "main" {
name = "main-waf"
scope = "REGIONAL"
description = "WAF untuk proteksi web server"
default_action {
allow {}
}
rule {
name = "AWS-BlockSQLi"
priority = 1
statement {
managed_rule_group_statement {
vendor_name = "AWS"
name = "AWSManagedRulesSQLiRuleSet"
}
}
action {
block {}
}
visibility_config {
cloudwatch_metrics_enabled = true
metric_name = "BlockSQLi"
sampled_requests_enabled = true
}
}
visibility_config {
cloudwatch_metrics_enabled = true
metric_name = "MainWAF"
sampled_requests_enabled = true
}
}
# variables.tf
variable "aws_region" {
description = "AWS region"
type = string
default = "ap-southeast-1"
}
variable "admin_cidr" {
description = "CIDR untuk akses admin"
type = string
default = "10.0.0.0/24"
}
Terraform Commands
# Init - download providers
terraform init
# Format kode
terraform fmt
# Validasi sintaks
terraform validate
# Lihat plan perubahan
terraform plan -out=tfplan
# Apply perubahan
terraform apply tfplan
# Destroy semua resource
terraform destroy
7. Contoh Implementasi Nyata
IR Automation Pipeline - Auto Response Insiden
#!/usr/bin/env python3
# ir-automation.py
# Incident Response Automation - deteksi, analisis, response, notifikasi
import os
import sys
import json
import logging
import subprocess
import requests
from datetime import datetime
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(message)s",
handlers=[
logging.FileHandler("/var/log/ir-automation.log"),
logging.StreamHandler()
]
)
logger = logging.getLogger(__name__)
class IncidentResponseAutomation:
"""Pipeline IR otomatis: deteksi -> analisis -> containment -> notify."""
def __init__(self, config_file="ir-config.json"):
with open(config_file) as f:
self.config = json.load(f)
self.siem_url = self.config["siem_url"]
self.siem_key = self.config["siem_key"]
self.webhook_url = self.config["webhook_url"]
def detect_incidents(self):
"""Poll SIEM untuk alert baru severity critical/high."""
headers = {"Authorization": f"Bearer {self.siem_key}"}
response = requests.get(
f"{self.siem_url}/api/alerts",
headers=headers,
params={
"severity": "critical,high",
"status": "new",
"limit": 10
},
timeout=30
)
response.raise_for_status()
return response.json()
def analyze_ip(self, ip_address):
"""Analisis IP dengan WHOIS dan threat intel."""
result = {"ip": ip_address, "reputation": "unknown"}
try:
# WHOIS lookup
whois_cmd = subprocess.run(
["whois", ip_address],
capture_output=True, text=True, timeout=10
)
if "NetRange" in whois_cmd.stdout:
result["asn"] = subprocess.run(
["grep", "OriginAS"],
input=whois_cmd.stdout, capture_output=True, text=True
).stdout.strip()
# Threat intel check (mock)
result["threat_score"] = 75
result["known_malicious"] = True
except Exception as e:
logger.error(f"Gagal analisis IP {ip_address}: {e}")
return result
def contain_ip(self, ip_address):
"""Containment - block IP di firewall dan update SIEM."""
actions = []
# Block via iptables
try:
subprocess.run(
["iptables", "-A", "INPUT", "-s", ip_address, "-j", "DROP"],
check=True, timeout=5
)
actions.append("iptables_block")
logger.info(f"[CONTAIN] IP {ip_address} blocked via iptables")
except subprocess.CalledProcessError as e:
logger.error(f"Gagal block iptables: {e}")
# Block via fail2ban jika ada
try:
subprocess.run(
["fail2ban-client", "set", "sshd", "banip", ip_address],
check=True, timeout=5
)
actions.append("fail2ban_block")
except Exception:
pass
return actions
def notify_team(self, incident, analysis, actions):
"""Kirim notifikasi ke Telegram/Slack."""
message = {
"text": (
f"🚨 *IR Automation Alert*\n\n"
f"*Alert:* {incident.get('title', 'N/A')}\n"
f"*Severity:* {incident.get('severity', 'N/A')}\n"
f"*Source IP:* {incident.get('source_ip', 'N/A')}\n"
f"*Threat Score:* {analysis.get('threat_score', 'N/A')}\n"
f"*Actions Taken:* {', '.join(actions)}\n"
f"*Timestamp:* {datetime.utcnow().isoformat()}"
)
}
try:
requests.post(self.webhook_url, json=message, timeout=5)
logger.info("Notifikasi terkirim")
except Exception as e:
logger.error(f"Gagal kirim notifikasi: {e}")
def run(self):
"""IR pipeline utama."""
logger.info("=== IR Automation Pipeline Start ===")
incidents = self.detect_incidents()
logger.info(f"Ditemukan {len(incidents)} incidents baru")
for incident in incidents:
ip = incident.get("source_ip")
if not ip:
continue
logger.info(f"Memproses incident dari IP: {ip}")
# Analisis
analysis = self.analyze_ip(ip)
# Containment
actions = self.contain_ip(ip)
# Update status di SIEM
headers = {"Authorization": f"Bearer {self.siem_key}"}
requests.patch(
f"{self.siem_url}/api/alerts/{incident['id']}",
headers=headers,
json={"status": "contained"}
)
# Notify
self.notify_team(incident, analysis, actions)
logger.info("=== IR Automation Pipeline Complete ===")
if __name__ == "__main__":
ir = IncidentResponseAutomation("/etc/ir-automation/ir-config.json")
ir.run()
Auto-Scan Pipeline - Nmap + Nuclei + Report
#!/bin/bash
# auto-scan-pipeline.sh
# Pipeline scanning otomatis: subnet discovery -> port scan -> vuln scan -> report
set -euo pipefail
# Konfigurasi
SUBNET="10.0.0.0/24"
SCAN_DIR="/scans/$(date +%Y%m%d_%H%M%S)"
NMAP_OPTIONS="-sS -sV -O --top-ports 1000"
NUCLEI_OPTIONS="-severity critical,high,medium"
WEBHOOK_URL="https://hooks.slack.com/services/T00/B00/xxxx"
# Notifikasi function
notify() {
local message="$1"
curl -s -X POST -H "Content-Type: application/json" \
-d "{\"text\": \"$message\"}" "$WEBHOOK_URL" >/dev/null
}
mkdir -p "$SCAN_DIR"
echo "[SCAN] Pipeline dimulai: $(date)"
notify "🔄 Scan pipeline dimulai untuk subnet $SUBNET"
# Phase 1: Host Discovery
echo "[PHASE 1] Host discovery..."
nmap -sn $SUBNET -oG "$SCAN_DIR/hosts.gnmap"
grep "Status: Up" "$SCAN_DIR/hosts.gnmap" | awk '{print $2}' > "$SCAN_DIR/live_hosts.txt"
LIVE_HOSTS=$(wc -l < "$SCAN_DIR/live_hosts.txt")
echo "[PHASE 1] Ditemukan $LIVE_HOSTS host hidup"
# Phase 2: Port Scan
echo "[PHASE 2] Port scan..."
nmap $NMAP_OPTIONS \
-iL "$SCAN_DIR/live_hosts.txt" \
-oA "$SCAN_DIR/portscan" 2>&1 | tail -5
# Phase 3: Vulnerability Scan dengan Nuclei
echo "[PHASE 3] Vulnerability scan..."
if command -v nuclei &>/dev/null; then
nuclei -l "$SCAN_DIR/live_hosts.txt" \
$NUCLEI_OPTIONS \
-o "$SCAN_DIR/nuclei_results.txt" 2>&1 | tail -5
fi
# Phase 4: Generate Report
echo "[PHASE 4] Generate report..."
{
echo "=== Security Scan Report ==="
echo "Date: $(date)"
echo "Subnet: $SUBNET"
echo "Hosts Found: $LIVE_HOSTS"
echo ""
echo "=== Open Ports ==="
grep "/open/" "$SCAN_DIR/portscan.gnmap" || echo "No open ports found"
echo ""
echo "=== Vulnerabilities ==="
if [ -f "$SCAN_DIR/nuclei_results.txt" ]; then
cat "$SCAN_DIR/nuclei_results.txt"
fi
} > "$SCAN_DIR/report.txt"
# Notify
VULN_COUNT=0
[ -f "$SCAN_DIR/nuclei_results.txt" ] && VULN_COUNT=$(wc -l < "$SCAN_DIR/nuclei_results.txt")
notify "✅ Scan selesai!
• Host: $LIVE_HOSTS
• Vulnerabilities: $VULN_COUNT
• Report: $SCAN_DIR/report.txt"
echo "[DONE] Pipeline selesai: $(date)"
echo "[DONE] Report: $SCAN_DIR/report.txt"
Automation Update - Auto-Patch Security
#!/bin/bash
# auto-update-security.sh
# Update otomatis dengan safety checks
# Jadwalkan via cron: 0 3 * * 0 /usr/local/bin/auto-update-security.sh
set -euo pipefail
LOG_FILE="/var/log/auto-update-security.log"
BACKUP_DIR="/backups/pre-update"
exec 1>> "$LOG_FILE" 2>&1
echo "[$(date)] === Auto Security Update Dimulai ==="
# 1. Backup konfigurasi
mkdir -p "$BACKUP_DIR/$(date +%Y%m%d)"
dpkg --get-selections > "$BACKUP_DIR/$(date +%Y%m%d)/packages.list"
cp -r /etc "$BACKUP_DIR/$(date +%Y%m%d)/etc"
# 2. Update package list
apt-get update -qq
echo "[$(date)] Package list updated"
# 3. Cek security updates saja
SECURITY_UPDATES=$(apt-get list --upgradable 2>/dev/null | grep -i security || true)
if [ -z "$SECURITY_UPDATES" ]; then
echo "[$(date)] Tidak ada security update"
echo "[$(date)] === Selesai ==="
exit 0
fi
echo "[$(date)] Security updates ditemukan:"
echo "$SECURITY_UPDATES"
# 4. Apply security updates
apt-get upgrade -y 2>&1 | tail -5
# 5. Cek apakah perlu reboot
if [ -f /var/run/reboot-required ]; then
echo "[$(date)] Reboot required - menjadwalkan ulang"
# Notifikasi admin
curl -s -X POST -H "Content-Type: application/json" \
-d '{"text":"⚠️ Reboot diperlukan setelah security update"}' \
"$WEBHOOK_URL" 2>/dev/null || true
fi
echo "[$(date)] === Auto Security Update Selesai ==="
8. Error Handling & Logging
Best Practice Error Handling
#!/bin/bash
# error-handling-template.sh
set -euo pipefail # Exit on error, undefined vars, pipe failures
IFS=$'\n\t' # Safe IFS
# Trap untuk cleanup
cleanup() {
local exit_code=$?
echo "[$(date)] Script exit dengan code: $exit_code"
# Hapus temporary files
rm -f /tmp/scan_temp_*.txt
exit $exit_code
}
trap cleanup EXIT INT TERM
# Fungsi logging
log_info() { echo "[INFO] $(date): $*"; }
log_warn() { echo "[WARN] $(date): $*"; }
log_error() { echo "[ERROR] $(date): $*" >&2; }
Notification - Kirim ke Telegram
#!/bin/bash
# send-telegram.sh
# Kirim notifikasi ke Telegram bot
TELEGRAM_BOT_TOKEN="${TELEGRAM_BOT_TOKEN:-}"
TELEGRAM_CHAT_ID="${TELEGRAM_CHAT_ID:-}"
send_telegram() {
local message="$1"
if [ -z "$TELEGRAM_BOT_TOKEN" ] || [ -z "$TELEGRAM_CHAT_ID" ]; then
echo "[WARN] Telegram credentials tidak dikonfigurasi"
return 1
fi
curl -s -X POST \
"https://api.telegram.org/bot$TELEGRAM_BOT_TOKEN/sendMessage" \
-d "chat_id=$TELEGRAM_CHAT_ID" \
-d "text=$message" \
-d "parse_mode=Markdown" >/dev/null
echo "[NOTIFY] Telegram terkirim"
}
# Contoh penggunaan dalam script
# send_telegram "✅ Backup selesai: $(date)"
# send_telegram "🚨 Alert: Brute force terdeteksi dari 10.0.0.5"
Notification - Kirim ke Slack
#!/usr/bin/env python3
# send-slack.py
# Kirim notifikasi ke Slack webhook
import requests
import sys
import json
def send_slack(message, webhook_url=None, color="good"):
"""Kirim notifikasi ke Slack via webhook."""
if not webhook_url:
webhook_url = "https://hooks.slack.com/services/xxx/xxx/xxx"
payload = {
"attachments": [{
"color": color,
"text": message,
"mrkdwn_in": ["text"]
}]
}
response = requests.post(webhook_url, json=payload)
return response.status_code == 200
if __name__ == "__main__":
message = sys.argv[1] if len(sys.argv) > 1 else "No message"
send_slack(message)
9. Best Practices
Automation Script Checklist
- Gunakan
set -euo pipefaildi setiap Bash script - Implementasi lock file - cegah multiple instance
- Logging ke file - dengan timestamp untuk audit
- Error handling - tangkap dan laporkan semua error
- Notification - kirim alert saat gagal atau sukses
- Testing - selalu jalankan dry run sebelum production
- Backup - backup konfigurasi sebelum perubahan
- Idempotent - script aman dijalankan berulang kali
- Configuration - pisahkan konfigurasi dari kode
Keamanan Script
- Jangan hardcode credentials - gunakan environment variable atau vault
- Validasi input - semua parameter harus dicek
- Least privilege - jalankan script dengan user minimal
- Audit trail - log semua aksi untuk forensik
- Encrypt secrets - jangan simpan password di plaintext