Программирование и скриптинг для кибербезопасности: часть 4 — кросс-платформенные воркфлоу мониторинга

from __future__ import annotations

import argparse
import sys
from pathlib import Path
from typing import Iterable, Iterator

DEFAULT_LOG_CANDIDATES: tuple[str, ...] = (
    "accesos.log",
    "access.log",
    "/var/log/apache2/access.log",
    "/var/log/httpd/access_log",
    "/var/log/nginx/access.log",
)

DEFAULT_TARGET = "sudo"
DEFAULT_OUTPUT = "reporte_sudo.txt"


def parse_args(argv: Iterable[str]) -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Filter an access log, find matches, and generate a report.",
    )
    parser.add_argument(
        "--log",
        dest="log_path",
        type=Path,
        default=None,
        help="Path to the log file (auto-detects common locations by default).",
    )
    parser.add_argument(
        "--target",
        dest="target",
        default=DEFAULT_TARGET,
        help="Literal string to search for in each line.",
    )
    parser.add_argument(
        "--out",
        dest="out_path",
        type=Path,
        default=Path(DEFAULT_OUTPUT),
        help="Report file to generate.",
    )
    parser.add_argument(
        "--encoding",
        dest="encoding",
        default="utf-8",
        help="Encoding used for reading and writing (default: utf-8).",
    )
    parser.add_argument(
        "--ignore-case",
        dest="ignore_case",
        action="store_true",
        help="Perform a case-insensitive search.",
    )
    parser.add_argument(
        "--dry-run",
        dest="dry_run",
        action="store_true",
        help="Show statistics without generating the output file.",
    )
    return parser.parse_args(list(argv))


def auto_detect_log_path() -> Path:
    for candidate in DEFAULT_LOG_CANDIDATES:
        path = Path(candidate)
        if path.exists():
            return path
    # Fall back to the first option (keeps behaviour predictable for the assignment)
    return Path(DEFAULT_LOG_CANDIDATES[0])


def read_matching_lines(
    ruta: Path,
    needle: str,
    *,
    encoding: str,
    ignore_case: bool,
) -> Iterator[str]:
    if not ruta.exists():
        raise FileNotFoundError(f"Log file not found: {ruta}")

    comparator = (lambda text: text.lower()) if ignore_case else (lambda text: text)
    target_cmp = comparator(needle)

    with ruta.open("r", encoding=encoding, errors="replace") as fh:
        for raw_line in fh:
            line = raw_line.rstrip("\n")
            if target_cmp in comparator(line):
                yield line


def write_report(
    ruta_out: Path,
    hallazgos: Iterable[str],
    *,
    encoding: str,
    target: str,
) -> int:
    total = 0
    if ruta_out.parent and not ruta_out.parent.exists():
        ruta_out.parent.mkdir(parents=True, exist_ok=True)
    with ruta_out.open("w", encoding=encoding, errors="replace") as fh:
        for line in hallazgos:
            fh.write(line + "\n")
            total += 1
        fh.write(f"\nTotal occurrences of '{target}': {total}\n")
    return total


def main(argv: Iterable[str] | None = None) -> int:
    ns = parse_args(argv or sys.argv[1:])

    log_path = ns.log_path or auto_detect_log_path()
    try:
        matches_iter = read_matching_lines(
            log_path,
            ns.target,
            encoding=ns.encoding,
            ignore_case=ns.ignore_case,
        )
    except FileNotFoundError as exc:
        print(f"[ERROR] {exc}")
        return 1
    except PermissionError as exc:
        print(f"[ERROR] Permission denied reading {log_path}: {exc}")
        return 2
    except OSError as exc:
        print(f"[ERROR] Failed to read {log_path}: {exc}")
        return 3

    if ns.dry_run:
        total = sum(1 for _ in matches_iter)
        print(
            f"[INFO] Matches found: {total} — dry-run mode, no report generated."
        )
        return 0

    try:
        total_written = write_report(
            ns.out_path,
            matches_iter,
            encoding=ns.encoding,
            target=ns.target,
        )
    except OSError as exc:
        print(f"[ERROR] Unable to write report '{ns.out_path}': {exc}")
        return 4

    print(
        f"[OK] Report generated: {ns.out_path} — Occurrences: {total_written} — Source: {log_path}"
    )
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

python3 log_monitor.py
python3 log_monitor.py --log /var/log/nginx/access.log --ignore-case --dry-run

param(
    [int]$Cantidad = 50,
    [string]$RutaSalida = ".\eventos.csv"
)

try {
    Write-Host "[INFO] Fetching last $Cantidad events from the 'Security' log..."
    $eventos = Get-EventLog -LogName Security -Newest $Cantidad -ErrorAction Stop

    Write-Host "[INFO] Exporting to CSV: $RutaSalida"
    $eventos | Export-Csv -Path $RutaSalida -NoTypeInformation -Encoding UTF8

    if (Test-Path -Path $RutaSalida) {
        Write-Host "[OK] CSV generated successfully: $RutaSalida"
        exit 0
    } else {
        Write-Host "[ERROR] CSV not found after export." -ForegroundColor Red
        exit 1
    }
}
catch {
    Write-Host "[ERROR] Failed to retrieve or export events: $($_.Exception.Message)" -ForegroundColor Red
    exit 2
}

# From an elevated PowerShell prompt
.\eventos_seguridad.ps1
.\eventos_seguridad.ps1 -Cantidad 100 -RutaSalida C:\Logs\eventos.csv