Advanced Linux Systems Security: SSH, Debian Package Management, and Text Processing

# Enable post-quantum algorithms in sshd_config
KexAlgorithms sntrup761x25519-sha512@openssh.com,curve25519-sha256
HostKeyAlgorithms ssh-ed25519-cert-v01@openssh.com,ssh-ed25519
Ciphers chacha20-poly1305@openssh.com,aes256-gcm@openssh.com

# Verify quantum-resistant handshake
ssh -vv user@host 2>&1 | grep -E "kex: algorithm|cipher:"

# Detect potential RegreSSHion exploitation attempts
grep -E "signal handler|race condition|SIGALRM" /var/log/auth.log | \
  awk '{print $1,$2,$3,$11}' | \
  uniq -c | \
  sort -rn | \
  head -20

# Monitor for unusual sshd child processes
ps aux | grep -E "sshd.*\[priv\]" | \
  awk '{if($8 ~ /Z/) print "ZOMBIE DETECTED:", $0}'

# Production-grade sshd_config hardening
cat <<'EOF' > /etc/ssh/sshd_config.d/99-hardening.conf
# Post-quantum ready configuration
RequiredRSASize 3072
Protocol 2
PermitRootLogin no
MaxAuthTries 3
MaxSessions 10
ClientAliveInterval 300
ClientAliveCountMax 2

# Process separation for RegreSSHion mitigation
UsePrivilegeSeparation sandbox
StrictModes yes

# Audit logging
LogLevel VERBOSE
SyslogFacility AUTH
EOF

# Apply and verify configuration
sshd -t && systemctl reload sshd
ss -tlnp | grep :22

# Generate SSH Certificate Authority
ssh-keygen -t ed25519 -f /etc/ssh/ca/user_ca -C "User Certificate Authority"
ssh-keygen -t ed25519 -f /etc/ssh/ca/host_ca -C "Host Certificate Authority"

# Sign user certificate with 8-hour validity
ssh-keygen -s /etc/ssh/ca/user_ca \
  -I "developer@company.com" \
  -n alice,deploy-role \
  -V +8h \
  -z $RANDOM \
  /home/alice/.ssh/id_ed25519.pub

# Sign host certificate for server fleet
for host in web{001..100}.prod; do
  ssh-keygen -s /etc/ssh/ca/host_ca \
    -I "$host.company.com" \
    -h \
    -n "$host.company.com,$host" \
    -V +365d \
    /etc/ssh/ssh_host_ed25519_key.pub
done

# Configure clients to trust CA
echo "@cert-authority *.company.com $(cat /etc/ssh/ca/host_ca.pub)" \
  >> ~/.ssh/known_hosts

# Configure servers to trust user CA
echo "TrustedUserCAKeys /etc/ssh/ca/user_ca.pub" >> /etc/ssh/sshd_config

# Lambda function for certificate issuance
import boto3
import json
from datetime import datetime, timedelta
import subprocess

def issue_certificate(event, context):
    # Verify OAuth token
    token = verify_oauth_token(event['headers']['Authorization'])

    # Generate certificate with temporal restrictions
    validity = "+8h" if token['role'] == 'developer' else "+5m"

    result = subprocess.run([
        'ssh-keygen', '-s', '/tmp/ca_key',
        '-I', f"{token['email']}-{datetime.now().isoformat()}",
        '-n', token['principals'],
        '-V', validity,
        '-z', str(random.randint(1, 999999)),
        '/tmp/user_key.pub'
    ], capture_output=True)

    # Log to CloudTrail
    boto3.client('cloudtrail').put_events(
        Records=[{
            'eventTime': datetime.now(),
            'eventName': 'SSHCertificateIssued',
            'userIdentity': {'principalId': token['email']},
            'requestParameters': {'validity': validity}
        }]
    )

    return {
        'statusCode': 200,
        'body': json.dumps({'certificate': result.stdout.decode()})
    }

# Advanced multiplexing configuration
cat <<'EOF' >> ~/.ssh/config
# Global multiplexing settings
ControlMaster auto
ControlPath ~/.ssh/control/%r@%h:%p
ControlPersist 10m

# Per-host optimization
Host *.prod.internal
  # Use fastest cipher for internal network
  Ciphers aes128-gcm@openssh.com
  # Disable compression on fast networks
  Compression no
  # Enable multiplexing
  ControlMaster auto

Host *.remote.satellite
  # ChaCha20 for systems without AES-NI
  Ciphers chacha20-poly1305@openssh.com
  # Enable compression for slow links
  Compression yes
  # Aggressive keepalive for unreliable connections
  ServerAliveInterval 30
  ServerAliveCountMax 10
EOF

# Performance benchmarking script
#!/bin/bash
for cipher in aes128-gcm aes256-gcm chacha20-poly1305; do
  echo "Testing $cipher..."
  time ssh -c "${cipher}@openssh.com" server \
    "dd if=/dev/zero bs=1M count=100" > /dev/null
done

# Monitor multiplexing effectiveness
lsof ~/.ssh/control/* | grep ESTABLISHED | wc -l

# Analyze package database state
#!/bin/bash
DB="/var/lib/dpkg/status"

# Parse status database
awk '/^Package:/ {pkg=$2}
     /^Status:/ {
       if ($4 != "installed") {
         print pkg, $2, $3, $4
       }
     }' $DB | column -t

# Recover from interrupted installation
dpkg --audit  # List problematic packages
dpkg --configure -a  # Complete configuration
apt-get install -f  # Fix dependencies

# Deep inspection of package state
strace -e openat dpkg -l nginx 2>&1 | grep -E "status|available"

# Database integrity check
debsums -c  # Verify installed files
debsums -g  # Generate missing checksums

# Backup before recovery attempt
cp -a /var/lib/dpkg /var/lib/dpkg.backup.$(date +%Y%m%d)

# Fix common corruption patterns
cat <<'SCRIPT' > /tmp/fix_dpkg.sh
#!/bin/bash
# Fix status file corruption
if ! grep -q "^Package: dpkg$" /var/lib/dpkg/status; then
  echo "CRITICAL: dpkg entry missing from status!"
  # Restore from available
  awk '/^Package: dpkg$/,/^$/' /var/lib/dpkg/available >> /var/lib/dpkg/status
fi

# Fix half-configured packages
for pkg in $(dpkg -l | grep "^iF" | awk '{print $2}'); do
  echo "Fixing half-configured: $pkg"
  dpkg --configure "$pkg" || apt-get install --reinstall "$pkg"
done

# Rebuild package cache
apt-get update
apt-cache gencaches
SCRIPT

bash /tmp/fix_dpkg.sh

#!/usr/bin/env python3
"""APT dependency resolver visualization"""

import apt
import networkx as nx
import matplotlib.pyplot as plt

def build_dependency_graph(package_name):
    cache = apt.Cache()
    graph = nx.DiGraph()

    def add_deps(pkg, level=0, max_level=3):
        if level > max_level:
            return

        pkg_obj = cache[pkg] if pkg in cache else None
        if not pkg_obj:
            return

        candidate = pkg_obj.candidate
        if not candidate:
            return

        # Add node with metadata
        graph.add_node(pkg,
                      size=candidate.size,
                      version=candidate.version)

        # Process dependencies
        for dep_type in ['Depends', 'Recommends', 'Suggests']:
            deps = candidate.get_dependencies(dep_type)
            for dep in deps:
                for base_dep in dep:
                    dep_name = base_dep.name
                    graph.add_edge(pkg, dep_name,
                                 relation=dep_type,
                                 version=base_dep.version)
                    add_deps(dep_name, level + 1)

    add_deps(package_name)
    return graph

# Analyze nginx dependencies
graph = build_dependency_graph('nginx')
print(f"Dependency tree: {len(graph.nodes)} packages")
print(f"Critical path: {nx.dag_longest_path(graph)}")

# Find potential conflicts
for node in graph.nodes():
    conflicts = [n for n in graph.nodes()
                 if 'Conflicts' in graph.get_edge_data(node, n, {})]
    if conflicts:
        print(f"Conflict detected: {node} -> {conflicts}")

# Modern repository setup (post apt-key)
#!/bin/bash
REPO_NAME="internal"
KEY_URL="https://packages.company.com/signing-key.asc"

# Download and verify key
wget -qO- "$KEY_URL" | gpg --dearmor | \
  sudo tee /etc/apt/keyrings/${REPO_NAME}.gpg > /dev/null

# Configure repository with signed-by
cat <<EOF | sudo tee /etc/apt/sources.list.d/${REPO_NAME}.list
deb [arch=amd64 signed-by=/etc/apt/keyrings/${REPO_NAME}.gpg] \
  https://packages.company.com/debian bullseye main
EOF

# Create local repository with reprepro
cat <<'EOF' > /srv/repo/conf/distributions
Origin: Company Internal
Label: Company Internal
Codename: bullseye
Architectures: amd64 arm64
Components: main
Description: Internal Package Repository
SignWith: ABCD1234
EOF

# Add packages with automatic signing
reprepro -b /srv/repo includedeb bullseye /tmp/*.deb

# Generate repository statistics
reprepro -b /srv/repo stats

/* Simplified Boyer-Moore implementation showing skip logic */
#include <string.h>
#include <stdio.h>

void compute_skip_table(const char *pattern, int *skip, int pattern_len) {
    // Initialize all characters to pattern length
    for (int i = 0; i < 256; i++) {
        skip[i] = pattern_len;
    }

    // Set skip values for characters in pattern
    for (int i = 0; i < pattern_len - 1; i++) {
        skip[(unsigned char)pattern[i]] = pattern_len - i - 1;
    }
}

int boyer_moore_search(const char *text, const char *pattern) {
    int text_len = strlen(text);
    int pattern_len = strlen(pattern);
    int skip[256];

    compute_skip_table(pattern, skip, pattern_len);

    int i = pattern_len - 1;
    while (i < text_len) {
        int j = pattern_len - 1;
        int k = i;

        while (j >= 0 && text[k] == pattern[j]) {
            j--;
            k--;
        }

        if (j < 0) {
            return k + 1;  // Pattern found
        }

        // Skip based on bad character
        i += skip[(unsigned char)text[i]];
    }

    return -1;  // Pattern not found
}

#!/bin/bash
# Performance benchmark: Boyer-Moore advantage

# Generate test file
perl -e 'print "random text " x 1000000' > test.txt
echo "authentication_failure_critical_security_event" >> test.txt

# Short pattern (slower)
time grep -c "auth" test.txt

# Long pattern (faster due to Boyer-Moore)
time grep -c "authentication_failure_critical_security_event" test.txt

# Demonstrate LC_ALL=C optimization
time grep -c "pattern" test.txt
time LC_ALL=C grep -c "pattern" test.txt

# Compare implementations
for tool in grep egrep "grep -F" rg ag; do
  echo "Testing: $tool"
  time $tool "authentication" test.txt > /dev/null
done

# PCRE lookaround for security analysis
#!/bin/bash

# Find SQL injection attempts not in comments
pcre2grep -M '(?<!--).*\bUNION\s+SELECT\b' access.log

# Extract passwords that aren't expired
pcre2grep -o 'password:\s*\K[^:]+(?!.*expired)' /etc/shadow.backup

# Complex log analysis with named groups
pcre2grep -o '(?<time>\d{2}:\d{2}:\d{2}).*?(?<ip>\d+\.\d+\.\d+\.\d+).*?(?<status>\d{3})' \
  access.log | \
  jq -R 'split(" ") | {time: .[0], ip: .[1], status: .[2]}'

# Performance-safe pattern with atomic groups
# Prevents catastrophic backtracking
pcre2grep '(?>\w+)*@(?>\w+\.)+\w+' emails.txt

# POSIX equivalent requires multiple passes
grep -E '\w+@\w+\.\w+' emails.txt | \
  grep -v '\.\.\.' | \
  grep -E '@.+\..+\.'

# Ripgrep for development workflows
rg --type-add 'config:*.{json,yaml,toml}' \
   --type config \
   --json \
   'api_key|password|secret' | \
   jq '.data.lines.text'

# Silver Searcher for code analysis
ag --stats \
   --ignore-dir=vendor \
   --file-search-regex='.*\.(go|rs)$' \
   'unsafe|panic'

# Parallel grep for distributed search
parallel -j+0 --slf /etc/nodes.txt \
  "grep -l 'ERROR.*CRITICAL' /var/log/app.log" | \
  sort | uniq

# Performance comparison script
#!/bin/bash
PATTERN="authentication.*failure"
LOGFILE="/var/log/auth.log"

hyperfine --warmup 3 \
  "grep -E '$PATTERN' $LOGFILE" \
  "rg '$PATTERN' $LOGFILE" \
  "ag '$PATTERN' $LOGFILE" \
  "awk '/$PATTERN/' $LOGFILE"

# Production-grade Ansible playbook
---
- name: Secure Package Deployment Pipeline
  hosts: production
  become: yes
  vars:
    package_repo: "https://packages.internal"
    ssh_ca_pub: "{{ vault_ssh_ca_public_key }}"

  tasks:
    - name: Configure SSH certificate authority
      copy:
        content: "{{ ssh_ca_pub }}"
        dest: /etc/ssh/trusted_user_ca
        mode: '0644'
      notify: reload_sshd

    - name: Add internal package repository
      block:
        - name: Download repository key
          get_url:
            url: "{{ package_repo }}/signing-key.asc"
            dest: /etc/apt/keyrings/internal.asc

        - name: Configure repository
          template:
            src: internal.list.j2
            dest: /etc/apt/sources.list.d/internal.list

    - name: Security audit before deployment
      shell: |
        debsecan --format json | \
        jq '.[] | select(.severity == "high" or .severity == "critical")'
      register: vulnerabilities
      failed_when: vulnerabilities.stdout != "[]"

    - name: Deploy application package
      apt:
        name: "{{ app_package }}"
        state: present
        update_cache: yes
      register: deployment

    - name: Verify deployment integrity
      shell: |
        debsums {{ app_package }} | grep -v OK$ || true
      register: integrity_check
      failed_when: integrity_check.stdout != ""

    - name: Security scan post-deployment
      shell: |
        # Check for suspicious SSH activity
        grep -E "Accepted publickey.*{{ app_user }}" /var/log/auth.log | \
        tail -10

        # Verify no unexpected services
        ss -tlnp | grep -v -E ":(22|443|8080)"
      register: security_scan

  handlers:
    - name: reload_sshd
      systemd:
        name: sshd
        state: reloaded

#!/usr/bin/env python3
"""Real-time security monitoring with pattern correlation"""

import re
import time
import subprocess
from collections import defaultdict
from datetime import datetime, timedelta
import json

class SecurityMonitor:
    def __init__(self):
        self.patterns = {
            'ssh_brute': r'Failed password.*from (?P<ip>\d+\.\d+\.\d+\.\d+)',
            'ssh_success': r'Accepted publickey for (?P<user>\w+).*from (?P<ip>\d+\.\d+\.\d+\.\d+)',
            'sudo_exec': r'sudo:.*COMMAND=(?P<cmd>.*)',
            'package_install': r'apt\[\d+\]:.*installed (?P<package>[\w\-]+)',
            'lateral_movement': r'Connection from (?P<src_ip>\d+\.\d+\.\d+\.\d+).*to (?P<dst_ip>\d+\.\d+\.\d+\.\d+)'
        }
        self.events = defaultdict(list)
        self.alerts = []

    def analyze_logs(self, log_file='/var/log/auth.log'):
        """Stream log analysis with pattern matching"""

        # Tail log file for real-time analysis
        proc = subprocess.Popen(
            ['tail', '-F', log_file],
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True
        )

        for line in proc.stdout:
            self.process_line(line.strip())
            self.correlate_events()

    def process_line(self, line):
        """Extract security events from log lines"""

        for event_type, pattern in self.patterns.items():
            match = re.search(pattern, line)
            if match:
                event = {
                    'type': event_type,
                    'timestamp': datetime.now(),
                    'data': match.groupdict(),
                    'raw': line
                }
                self.events[event_type].append(event)

                # Check for immediate threats
                if event_type == 'ssh_brute':
                    self.check_brute_force(match.group('ip'))
                elif event_type == 'lateral_movement':
                    self.check_lateral_movement(event)

    def check_brute_force(self, ip):
        """Detect brute force attacks"""

        recent = [e for e in self.events['ssh_brute']
                 if e['data']['ip'] == ip
                 and e['timestamp'] > datetime.now() - timedelta(minutes=5)]

        if len(recent) > 10:
            self.alert(f"BRUTE FORCE: {ip} - {len(recent)} attempts in 5 minutes")
            self.block_ip(ip)

    def check_lateral_movement(self, event):
        """Detect potential lateral movement"""

        src_ip = event['data']['src_ip']

        # Check if this IP recently had successful authentication
        recent_auth = [e for e in self.events['ssh_success']
                      if e['data']['ip'] == src_ip
                      and e['timestamp'] > event['timestamp'] - timedelta(minutes=1)]

        if recent_auth:
            self.alert(f"LATERAL MOVEMENT: {src_ip} -> {event['data']['dst_ip']}")

    def correlate_events(self):
        """Advanced event correlation"""

        # Pattern: SSH login -> sudo -> package install -> outbound connection
        for ssh_event in self.events['ssh_success'][-10:]:
            user = ssh_event['data']['user']
            time_window = ssh_event['timestamp'] + timedelta(minutes=5)

            # Find related sudo commands
            sudo_events = [e for e in self.events['sudo_exec']
                          if e['timestamp'] > ssh_event['timestamp']
                          and e['timestamp'] < time_window]

            # Find package installations
            package_events = [e for e in self.events['package_install']
                             if e['timestamp'] > ssh_event['timestamp']
                             and e['timestamp'] < time_window]

            if sudo_events and package_events:
                self.alert(f"SUSPICIOUS PATTERN: {user} installed packages after SSH login")

    def block_ip(self, ip):
        """Automatically block malicious IPs"""

        subprocess.run([
            'iptables', '-A', 'INPUT',
            '-s', ip,
            '-j', 'DROP'
        ])

        # Add to fail2ban
        subprocess.run([
            'fail2ban-client', 'set', 'sshd',
            'banip', ip
        ])

    def alert(self, message):
        """Send security alerts"""

        alert = {
            'timestamp': datetime.now().isoformat(),
            'message': message,
            'severity': 'HIGH'
        }

        self.alerts.append(alert)
        print(f"[ALERT] {message}")

        # Send to SIEM
        subprocess.run([
            'logger', '-p', 'auth.crit',
            f"SECURITY: {message}"
        ])

# Run monitor
if __name__ == "__main__":
    monitor = SecurityMonitor()
    monitor.analyze_logs()

#!/bin/bash
# Predictive performance monitoring system

# SSH connection health monitoring
monitor_ssh_performance() {
    local host=$1
    local threshold=2  # seconds

    while true; do
        start=$(date +%s.%N)

        # Test SSH handshake time
        timeout 10 ssh -o BatchMode=yes \
                      -o ConnectTimeout=5 \
                      "$host" exit 2>/dev/null

        end=$(date +%s.%N)
        duration=$(echo "$end - $start" | bc)

        # Predictive analysis
        if (( $(echo "$duration > $threshold" | bc -l) )); then
            echo "[WARNING] SSH to $host slow: ${duration}s"

            # Collect diagnostics
            ssh "$host" "top -bn1 | head -20; netstat -s | grep -i drop"
        fi

        # Store metrics for trend analysis
        echo "$(date +%s),ssh_latency,$host,$duration" >> /var/log/metrics.csv

        sleep 60
    done
}

# Package operation performance tracking
monitor_package_performance() {
    # Pre-deployment baseline
    time apt-get update 2>&1 | tee /tmp/apt-update.log

    # Extract performance metrics
    grep -oP 'Fetched \K[0-9,]+' /tmp/apt-update.log | tr -d ','

    # Monitor dpkg database operations
    strace -T -e openat,read,write apt-get install -s nginx 2>&1 | \
        awk -F'<|>' '/^[^<]*<[0-9.]+>$/ {sum+=$2; count++}
             END {print "Avg syscall time:", sum/count}'
}

# Real-time text processing performance
monitor_grep_performance() {
    local logfile="/var/log/syslog"
    local patterns=("ERROR" "WARNING" "authentication" "failed.*password")

    for pattern in "${patterns[@]}"; do
        # Benchmark different implementations
        echo "Pattern: $pattern"

        # Standard grep
        time1=$(time (grep -c "$pattern" "$logfile" 2>/dev/null) 2>&1 | \
                grep real | awk '{print $2}')

        # Optimized with LC_ALL=C
        time2=$(time (LC_ALL=C grep -c "$pattern" "$logfile" 2>/dev/null) 2>&1 | \
                grep real | awk '{print $2}')

        # Ripgrep
        time3=$(time (rg -c "$pattern" "$logfile" 2>/dev/null) 2>&1 | \
                grep real | awk '{print $2}')

        echo "  grep: $time1, LC_ALL=C: $time2, ripgrep: $time3"
    done
}

# Predictive failure detection
predict_failures() {
    # Analyze historical metrics
    awk -F',' '$2=="ssh_latency" {print $4}' /var/log/metrics.csv | \
    python3 -c "
import sys
import numpy as np
from scipy import stats

data = [float(line.strip()) for line in sys.stdin]
if len(data) > 100:
    # Calculate trend
    x = np.arange(len(data))
    slope, intercept, r_value, p_value, std_err = stats.linregress(x, data)

    if slope > 0.01:  # Increasing latency trend
        print(f'WARNING: Performance degrading, slope={slope:.4f}')

        # Predict failure point
        failure_threshold = 5.0  # 5 seconds
        failure_point = (failure_threshold - intercept) / slope
        print(f'Predicted failure in {failure_point:.0f} measurements')
"
}

# Main monitoring loop
while true; do
    monitor_ssh_performance "prod-server-01" &
    monitor_package_performance
    monitor_grep_performance
    predict_failures

    sleep 300  # 5-minute intervals
done

#!/bin/bash
# Full exercise solution with learning progression

# Stage 1: Manual package creation
create_package_manually() {
    mkdir -p myapp/DEBIAN
    mkdir -p myapp/usr/local/bin

    cat > myapp/DEBIAN/control <<EOF
Package: myapp
Version: 1.0.0
Architecture: amd64
Maintainer: Student <student@example.com>
Description: Learning exercise application
Depends: curl, grep
EOF

    cat > myapp/usr/local/bin/myapp <<'EOF'
#!/bin/bash
echo "MyApp v1.0.0 - Secure Installation"
ssh-keygen -l -f /etc/ssh/ssh_host_ed25519_key.pub
dpkg -l | grep -c "^ii"
EOF

    chmod 755 myapp/usr/local/bin/myapp
    dpkg-deb --build myapp
}

# Stage 2: Repository creation with GPG signing
create_repository() {
    # Generate GPG key for repository
    gpg --batch --generate-key <<EOF
Key-Type: RSA
Key-Length: 4096
Name-Real: Package Repository
Name-Email: repo@example.com
Expire-Date: 1y
%no-protection
%commit
EOF

    # Export public key
    gpg --armor --export repo@example.com > repo-key.asc

    # Create repository structure
    mkdir -p repo/pool/main/m/myapp
    mkdir -p repo/dists/stable/main/binary-amd64

    # Sign and add package
    dpkg-sig --sign builder myapp.deb
    cp myapp.deb repo/pool/main/m/myapp/

    # Generate repository metadata
    cd repo
    apt-ftparchive packages pool > dists/stable/main/binary-amd64/Packages
    gzip -k dists/stable/main/binary-amd64/Packages

    apt-ftparchive release dists/stable > dists/stable/Release
    gpg --default-key repo@example.com \
        --armor --detach-sign \
        -o dists/stable/Release.gpg \
        dists/stable/Release
}

# Stage 3: SSH certificate-based access
setup_ssh_certificates() {
    # Generate CA
    ssh-keygen -t ed25519 -f repo_ca -N "" -C "Repository CA"

    # Configure SSH server for certificate auth
    cat >> /etc/ssh/sshd_config <<EOF
TrustedUserCAKeys /etc/ssh/repo_ca.pub
AuthorizedPrincipalsFile /etc/ssh/principals/%u
EOF

    # Create time-limited certificates for repo access
    issue_repo_certificate() {
        local user=$1
        local validity=$2

        ssh-keygen -s repo_ca \
            -I "$user-repo-access" \
            -n "repo-reader,repo-writer" \
            -V "$validity" \
            ~/.ssh/id_ed25519.pub
    }

    # Setup repository access over SSH
    cat > /usr/local/bin/ssh-apt-transport <<'EOF'
#!/bin/bash
# Custom APT transport for SSH repositories
ssh -o BatchMode=yes repo@repository.internal \
    "cat /srv/repo/$1"
EOF

    chmod +x /usr/local/bin/ssh-apt-transport
}

# Stage 4: Integration and automation
integrate_ci_cd() {
    cat > .gitlab-ci.yml <<'EOF'
stages:
  - build
  - sign
  - deploy

build:package:
  stage: build
  script:
    - dpkg-deb --build myapp
    - debsums -g myapp.deb
  artifacts:
    paths:
      - myapp.deb

sign:package:
  stage: sign
  script:
    - dpkg-sig --sign builder myapp.deb
    - gpg --verify myapp.deb.sig
  artifacts:
    paths:
      - myapp.deb
      - myapp.deb.sig

deploy:repository:
  stage: deploy
  script:
    - ssh-keygen -s $CA_KEY -I "ci-deploy" -n "repo-writer" -V +5m id_ed25519.pub
    - scp myapp.deb repo@repository.internal:/srv/repo/incoming/
    - ssh repo@repository.internal "process-incoming"
EOF
}

# Execute all stages
create_package_manually
create_repository
setup_ssh_certificates
integrate_ci_cd

#!/usr/bin/env python3
"""
Complete security audit system with pattern detection
Demonstrates integration of SSH, package, and text processing
"""

import subprocess
import json
import re
from datetime import datetime, timedelta
import paramiko
import apt
from typing import Dict, List, Tuple

class SecurityAuditor:
    def __init__(self, hosts: List[str]):
        self.hosts = hosts
        self.vulnerabilities = []
        self.patterns = self.load_patterns()

    def load_patterns(self) -> Dict[str, str]:
        """Load security patterns for grep analysis"""
        return {
            'sql_injection': r"(\bUNION\b.*\bSELECT\b|\bOR\b.*=|--|\#|\/\*)",
            'command_injection': r"(;|\||&&|\$\(|\`)",
            'path_traversal': r"(\.\./|\.\.\\|%2e%2e)",
            'failed_auth': r"Failed password|authentication failure",
            'privilege_escalation': r"sudo.*COMMAND|su\[.*\]:",
            'suspicious_download': r"(wget|curl).*\.(sh|py|exe|dll)",
            'port_scan': r"Connection from.*port [0-9]+ on.*port",
            'brute_force': r"Failed password.*([0-9]{1,3}\.){3}[0-9]{1,3}",
        }

    def audit_ssh_configuration(self, host: str) -> Dict:
        """Audit SSH configuration for security issues"""

        ssh = paramiko.SSHClient()
        ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())

        try:
            ssh.connect(host, username='auditor', key_filename='~/.ssh/id_ed25519')

            # Check SSH configuration
            stdin, stdout, stderr = ssh.exec_command('sshd -T')
            config = stdout.read().decode()

            issues = []

            # Security checks
            if 'permitrootlogin yes' in config.lower():
                issues.append({'severity': 'HIGH', 'issue': 'Root login enabled'})

            if 'passwordauthentication yes' in config.lower():
                issues.append({'severity': 'MEDIUM', 'issue': 'Password auth enabled'})

            if 'protocol 1' in config.lower():
                issues.append({'severity': 'CRITICAL', 'issue': 'SSH Protocol 1 enabled'})

            # Check for weak algorithms
            stdin, stdout, stderr = ssh.exec_command(
                "grep -E 'Ciphers|MACs|KexAlgorithms' /etc/ssh/sshd_config"
            )
            algorithms = stdout.read().decode()

            weak_ciphers = ['3des', 'arcfour', 'des']
            for cipher in weak_ciphers:
                if cipher in algorithms.lower():
                    issues.append({
                        'severity': 'HIGH',
                        'issue': f'Weak cipher: {cipher}'
                    })

            return {'host': host, 'ssh_issues': issues}

        finally:
            ssh.close()

    def audit_packages(self, host: str) -> Dict:
        """Audit installed packages for vulnerabilities"""

        ssh = paramiko.SSHClient()
        ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())

        try:
            ssh.connect(host, username='auditor', key_filename='~/.ssh/id_ed25519')

            # Get package list
            stdin, stdout, stderr = ssh.exec_command('dpkg -l | grep "^ii"')
            packages = stdout.read().decode()

            # Run debsecan
            stdin, stdout, stderr = ssh.exec_command('debsecan --format json')
            vulnerabilities = json.loads(stdout.read().decode())

            # Analyze critical vulnerabilities
            critical = [v for v in vulnerabilities
                       if v.get('severity') in ['high', 'critical']]

            # Check for suspicious packages
            suspicious_patterns = [
                'netcat', 'nmap', 'wireshark', 'metasploit',
                'aircrack', 'hashcat', 'hydra'
            ]

            suspicious = []
            for pattern in suspicious_patterns:
                if pattern in packages.lower():
                    suspicious.append(pattern)

            return {
                'host': host,
                'critical_vulns': len(critical),
                'total_vulns': len(vulnerabilities),
                'suspicious_packages': suspicious
            }

        finally:
            ssh.close()

    def analyze_logs(self, host: str) -> Dict:
        """Analyze logs for security patterns"""

        ssh = paramiko.SSHClient()
        ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())

        try:
            ssh.connect(host, username='auditor', key_filename='~/.ssh/id_ed25519')

            findings = {}

            for pattern_name, pattern in self.patterns.items():
                # Search in auth.log
                cmd = f"grep -c -E '{pattern}' /var/log/auth.log 2>/dev/null || echo 0"
                stdin, stdout, stderr = ssh.exec_command(cmd)
                auth_count = int(stdout.read().decode().strip())

                # Search in syslog
                cmd = f"grep -c -E '{pattern}' /var/log/syslog 2>/dev/null || echo 0"
                stdin, stdout, stderr = ssh.exec_command(cmd)
                sys_count = int(stdout.read().decode().strip())

                if auth_count > 0 or sys_count > 0:
                    findings[pattern_name] = {
                        'auth_log': auth_count,
                        'syslog': sys_count
                    }

                    # Get sample matches for critical patterns
                    if pattern_name in ['sql_injection', 'command_injection']:
                        cmd = f"grep -E '{pattern}' /var/log/*/access.log 2>/dev/null | head -5"
                        stdin, stdout, stderr = ssh.exec_command(cmd)
                        samples = stdout.read().decode().strip()
                        if samples:
                            findings[pattern_name]['samples'] = samples

            return {'host': host, 'log_findings': findings}

        finally:
            ssh.close()

    def generate_report(self) -> str:
        """Generate comprehensive security report"""

        report = []
        report.append("=" * 60)
        report.append("SECURITY AUDIT REPORT")
        report.append(f"Generated: {datetime.now().isoformat()}")
        report.append("=" * 60)

        for host in self.hosts:
            report.append(f"\nHost: {host}")
            report.append("-" * 40)

            # SSH audit
            ssh_audit = self.audit_ssh_configuration(host)
            if ssh_audit['ssh_issues']:
                report.append("SSH Configuration Issues:")
                for issue in ssh_audit['ssh_issues']:
                    report.append(f"  [{issue['severity']}] {issue['issue']}")

            # Package audit
            pkg_audit = self.audit_packages(host)
            report.append(f"\nPackage Vulnerabilities:")
            report.append(f"  Critical: {pkg_audit['critical_vulns']}")
            report.append(f"  Total: {pkg_audit['total_vulns']}")
            if pkg_audit['suspicious_packages']:
                report.append(f"  Suspicious: {', '.join(pkg_audit['suspicious_packages'])}")

            # Log analysis
            log_audit = self.analyze_logs(host)
            if log_audit['log_findings']:
                report.append("\nSecurity Pattern Detections:")
                for pattern, counts in log_audit['log_findings'].items():
                    report.append(f"  {pattern}: auth={counts['auth_log']}, sys={counts['syslog']}")
                    if 'samples' in counts:
                        report.append(f"    Sample: {counts['samples'][:100]}...")

        return "\n".join(report)

    def continuous_monitoring(self):
        """Run continuous security monitoring"""

        while True:
            report = self.generate_report()

            # Save report
            filename = f"security_audit_{datetime.now().strftime('%Y%m%d_%H%M%S')}.txt"
            with open(filename, 'w') as f:
                f.write(report)

            # Check for critical issues
            if "CRITICAL" in report:
                self.send_alert(report)

            # Wait before next scan
            time.sleep(3600)  # 1 hour

    def send_alert(self, report: str):
        """Send critical security alerts"""

        # Email alert
        subprocess.run([
            'mail', '-s', 'CRITICAL Security Alert',
            'security@example.com'
        ], input=report.encode())

        # Slack webhook
        webhook_url = "https://hooks.slack.com/services/XXX"
        subprocess.run([
            'curl', '-X', 'POST',
            '-H', 'Content-type: application/json',
            '--data', json.dumps({'text': 'Critical security issue detected!'}),
            webhook_url
        ])

# Execute audit
if __name__ == "__main__":
    hosts = ['web01.prod', 'web02.prod', 'db01.prod']
    auditor = SecurityAuditor(hosts)
    auditor.continuous_monitoring()