#!/usr/bin/env python

from __future__ import print_function

import os
import sys
import subprocess
import tempfile
import stat
import time
from datetime import datetime

LOG_DIR = "/var/duckdb"

if not os.path.exists(LOG_DIR):
    os.makedirs(LOG_DIR)

LOCK_FILE = "/var/duckdb/logs-ingest.lock"

def acquire_lock():
    if os.path.exists(LOCK_FILE):
        try:
            with open(LOCK_FILE, "r") as f:
                pid = int(f.read().strip())
            os.kill(pid, 0)
            print("[-] Error: Another instance is already running (PID %d). Exiting." % pid)
            sys.exit(0)
        except (ValueError, OSError):
            os.remove(LOCK_FILE)
    with open(LOCK_FILE, "w") as f:
        f.write(str(os.getpid()))

def release_lock():
    if os.path.exists(LOCK_FILE):
        os.remove(LOCK_FILE)

def get_logwatch_keep_days(default_days=30):
    settings_file = "/var/efw/logging/settings"
    default_file  = "/usr/lib/efw/logging/default"
    
    target_file = settings_file if os.path.exists(settings_file) else default_file

    if os.path.exists(target_file):
        try:
            with open(target_file, "r") as f:
                for line in f:
                    line = line.strip()
                    if line.startswith("LOGWATCH_KEEP="):
                        val = line.split("=", 1)[1].strip().strip('"').strip("'")
                        return int(val)
        except Exception as e:
            print("[!] Warning: Could not read retention setting from %s: %s" % (target_file, e))

    return default_days

acquire_lock()

print("[+] Starting log ingestion process...")
start_time = time.time()

CURRENT_YEAR    = datetime.now().year
DB_FIREWALL     = os.path.join(LOG_DIR, "firewall.db")
DB_MESSAGES     = os.path.join(LOG_DIR, "messages.db")
DB_SURICATA     = os.path.join(LOG_DIR, "suricata.db")

OFFSET_FILE      = os.path.join(LOG_DIR, "firewall.offset")
TMP_LOG          = os.path.join(LOG_DIR, "firewall_delta.log")
MSG_OFFSET_FILE  = os.path.join(LOG_DIR, "messages.offset")
TMP_MSG_LOG      = os.path.join(LOG_DIR, "messages_delta.log")
SURI_OFFSET_FILE = os.path.join(LOG_DIR, "suricata.offset")
TMP_SURI_LOG     = os.path.join(LOG_DIR, "suricata_delta.log")

GEOIP_BASE = "/usr/lib/geoip"

def find_geoip_csv_dir(base):
    candidates = []
    for dirpath, dirnames, _ in os.walk(base):
        for d in dirnames:
            if d.startswith("GeoLite2-Country-CSV_"):
                candidates.append(os.path.join(dirpath, d))
    return sorted(candidates, reverse=True)[0] if candidates else None

GEOIP_CSV_DIR = find_geoip_csv_dir(GEOIP_BASE)
GEOIP_BLOCKS_V4 = None
GEOIP_LOCATIONS = None
GEOIP_VERSION_TAG = "NONE"

if not GEOIP_CSV_DIR:
    print("[!] Warning: GeoLite2-Country-CSV directory not found under %s. Geolocation will be skipped." % GEOIP_BASE)
else:
    GEOIP_BLOCKS_V4 = os.path.join(GEOIP_CSV_DIR, "GeoLite2-Country-Blocks-IPv4.csv")
    GEOIP_LOCATIONS = os.path.join(GEOIP_CSV_DIR, "GeoLite2-Country-Locations-en.csv")
    GEOIP_VERSION_TAG = os.path.basename(GEOIP_CSV_DIR)
    print("[+] MaxMind GeoIP database identified: %s" % GEOIP_VERSION_TAG)

DUCKDB_BIN = "/usr/bin/duckdb"
if not os.path.isfile(DUCKDB_BIN) or not os.access(DUCKDB_BIN, os.X_OK):
    print("[-] Critical Error: duckdb not found at /usr/bin/duckdb", file=sys.stderr)
    release_lock()
    sys.exit(1)

DUCKDB_THREADS = max(1, os.cpu_count() // 2) if hasattr(os, "cpu_count") else 1
if not hasattr(os, "cpu_count"):
    try:
        import multiprocessing
        DUCKDB_THREADS = max(1, multiprocessing.cpu_count() // 2)
    except Exception:
        DUCKDB_THREADS = 1

def run_logtail(log_file, offset_file, output_file):
    with open(output_file, "w") as out:
        with open(os.devnull, "w") as devnull:
            try:
                subprocess.call(
                    ["logtail", "-f", log_file, "-o", offset_file],
                    stdout=out,
                    stderr=devnull,
                )
            except OSError as e:
                print("[-] Critical Error: 'logtail' binary not found or failed to execute (%s)." % e, file=sys.stderr)
                release_lock()
                sys.exit(1)

print("[+] Collecting new log entries using logtail...")
run_logtail("/var/log/firewall", OFFSET_FILE, TMP_LOG)
run_logtail("/var/log/messages", MSG_OFFSET_FILE, TMP_MSG_LOG)
run_logtail("/var/log/suricata/alerts.log", SURI_OFFSET_FILE, TMP_SURI_LOG)

fw_size   = os.path.getsize(TMP_LOG) if os.path.exists(TMP_LOG) else 0
msg_size  = os.path.getsize(TMP_MSG_LOG) if os.path.exists(TMP_MSG_LOG) else 0
suri_size = os.path.getsize(TMP_SURI_LOG) if os.path.exists(TMP_SURI_LOG) else 0

fw_status = -1
if fw_size > 0:
    print("[+] Importing Firewall log delta (%d bytes) into firewall.db..." % fw_size)
    SQL_FIREWALL = """
    PRAGMA threads=%(DUCKDB_THREADS)s;

    CREATE TABLE IF NOT EXISTS log_firewall (
        event_date TIMESTAMP,
        src_ip VARCHAR,
        dst_ip VARCHAR,
        dst_port INTEGER,
        interface VARCHAR,
        protocol VARCHAR,
        action VARCHAR,
        chain VARCHAR,
        src_country_code VARCHAR,
        src_country_name VARCHAR,
        dst_country_code VARCHAR,
        dst_country_name VARCHAR
    );

    WITH fw_lines AS (
        SELECT unnest(string_split_regex(content, '\\r?\\n')) AS line
        FROM read_text('%(TMP_LOG)s')
    ),
    new_fw AS (
        SELECT
            try_strptime('%(CURRENT_YEAR)s ' || substring(line, 1, 15), '%%Y %%b %%d %%H:%%M:%%S') AS event_date,
            string_split(string_split(line, 'SRC=')[2], ' ')[1] AS src_ip,
            string_split(string_split(line, 'DST=')[2], ' ')[1] AS dst_ip,
            CASE
                WHEN line LIKE '%%DPT=%%' THEN
                    TRY_CAST(string_split(string_split(line, 'DPT=')[2], ' ')[1] AS INTEGER)
                ELSE NULL
            END AS dst_port,
            CASE WHEN line LIKE '%%IN=%%'    THEN string_split(string_split(line, 'IN=')[2],    ' ')[1] ELSE NULL END AS interface,
            CASE WHEN line LIKE '%%PROTO=%%' THEN string_split(string_split(line, 'PROTO=')[2], ' ')[1] ELSE NULL END AS protocol,
            UPPER(COALESCE(
                regexp_extract(line, 'ulogd\\[\\d+\\]:\\s*([A-Z]+):([A-Z]+)', 2), 
                regexp_extract(line, 'ulogd\\[\\d+\\]:\\s*([A-Z]+):([A-Z]+)', 1), 
                'DROP'
            )) AS action,
            regexp_extract(line, 'ulogd\\[\\d+\\]:\\s*([A-Z]+)', 1) AS chain
        FROM fw_lines
        WHERE line IS NOT NULL
          AND line != ''
          AND line LIKE '%%SRC=%%'
    )
    INSERT INTO log_firewall (
        event_date, src_ip, dst_ip, dst_port, interface, protocol, action, chain,
        src_country_code, src_country_name, dst_country_code, dst_country_name
    )
    SELECT DISTINCT
        n.event_date, n.src_ip, n.dst_ip, n.dst_port, n.interface, n.protocol, n.action, n.chain,
        'PENDING' AS src_country_code, 'Processing...' AS src_country_name,
        'PENDING' AS dst_country_code, 'Processing...' AS dst_country_name
    FROM new_fw n
    WHERE n.event_date IS NOT NULL
      AND NOT EXISTS (
          SELECT 1 FROM log_firewall e
          WHERE e.event_date = n.event_date
            AND COALESCE(e.src_ip, '') = COALESCE(n.src_ip, '')
            AND COALESCE(e.dst_ip, '') = COALESCE(n.dst_ip, '')
            AND COALESCE(e.dst_port, 0) = COALESCE(n.dst_port, 0)
            AND COALESCE(e.protocol, '') = COALESCE(n.protocol, '')
            AND COALESCE(e.action, '') = COALESCE(n.action, '')
      );
    """ % {
        "TMP_LOG":        TMP_LOG,
        "CURRENT_YEAR":   CURRENT_YEAR,
        "DUCKDB_THREADS": DUCKDB_THREADS,
    }

    tmp_sql_fd, tmp_sql_path = tempfile.mkstemp(suffix=".sql")
    try:
        with os.fdopen(tmp_sql_fd, "w") as tmp_sql_file:
            tmp_sql_file.write(SQL_FIREWALL)
        with open(tmp_sql_path, "r") as sql_input:
            proc = subprocess.Popen([DUCKDB_BIN, DB_FIREWALL], stdin=sql_input, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
            _, stderr_out = proc.communicate()
            fw_status = proc.returncode
    finally:
        if os.path.exists(tmp_sql_path):
            os.remove(tmp_sql_path)
    if fw_status != 0:
        sys.stderr.write("[-] Firewall ingestion error:\n" + stderr_out.decode("utf-8", errors="replace"))
else:
    print("[+] No new Firewall logs to process.")

msg_status = -1
if msg_size > 0:
    print("[+] Importing OS log delta (%d bytes) into messages.db..." % msg_size)
    SQL_MESSAGES = """
    PRAGMA threads=%(DUCKDB_THREADS)s;

    CREATE TABLE IF NOT EXISTS log_messages (
        event_date TIMESTAMP,
        service VARCHAR,
        message VARCHAR
    );

    CREATE OR REPLACE TEMP TABLE temp_msg_lines AS
    SELECT unnest(string_split_regex(content, '\\r?\\n')) AS line
    FROM read_text('%(TMP_MSG_LOG)s')
    WHERE content IS NOT NULL AND content != '';

    WITH parsed AS (
        SELECT
            strptime('%(CURRENT_YEAR)s ' || regexp_replace(substring(line, 1, 15), '\\s+', ' ', 'g'), '%%Y %%b %%d %%H:%%M:%%S') AS event_date,
            regexp_extract(line, '^[A-Za-z]{3}\\s+\\d+\\s+\\d{2}:\\d{2}:\\d{2}\\s+[^\\s]+\\s+([^:\\[]+)', 1) AS service,
            regexp_extract(line, '^[A-Za-z]{3}\\s+\\d+\\s+\\d{2}:\\d{2}:\\d{2}\\s+[^\\s]+\\s+[^:]+:\\s*(.*)$', 1) AS message
        FROM temp_msg_lines
        WHERE line IS NOT NULL
          AND line != ''
          AND length(line) > 16
    )
    INSERT INTO log_messages (event_date, service, message)
    SELECT p.event_date, p.service, p.message
    FROM parsed p
    WHERE p.event_date IS NOT NULL
      AND NOT EXISTS (
          SELECT 1 FROM log_messages e
          WHERE e.event_date = p.event_date
            AND COALESCE(e.service, '') = COALESCE(p.service, '')
            AND COALESCE(e.message, '') = COALESCE(p.message, '')
      );
    """ % {
        "TMP_MSG_LOG":    TMP_MSG_LOG,
        "CURRENT_YEAR":   CURRENT_YEAR,
        "DUCKDB_THREADS": DUCKDB_THREADS,
    }

    tmp_sql_fd, tmp_sql_path = tempfile.mkstemp(suffix=".sql")
    try:
        with os.fdopen(tmp_sql_fd, "w") as tmp_sql_file:
            tmp_sql_file.write(SQL_MESSAGES)
        with open(tmp_sql_path, "r") as sql_input:
            proc = subprocess.Popen([DUCKDB_BIN, DB_MESSAGES], stdin=sql_input, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
            _, stderr_out = proc.communicate()
            msg_status = proc.returncode
    finally:
        if os.path.exists(tmp_sql_path):
            os.remove(tmp_sql_path)
    if msg_status != 0:
        sys.stderr.write("[-] Messages ingestion error:\n" + stderr_out.decode("utf-8", errors="replace"))
else:
    print("[+] No new system logs (messages) to process.")

suri_status = -1
if suri_size > 0:
    print("[+] Importing Suricata log delta (%d bytes) into suricata.db..." % suri_size)
    SQL_SURICATA = """
    PRAGMA threads=%(DUCKDB_THREADS)s;

    CREATE TABLE IF NOT EXISTS log_suricata (
        event_date TIMESTAMP,
        action VARCHAR,
        sid INTEGER,
        signature VARCHAR,
        classification VARCHAR,
        priority INTEGER,
        protocol VARCHAR,
        src_ip VARCHAR,
        src_port INTEGER,
        dst_ip VARCHAR,
        dst_port INTEGER,
        src_country_code VARCHAR,
        src_country_name VARCHAR,
        dst_country_code VARCHAR,
        dst_country_name VARCHAR
    );

    WITH suri_lines AS (
        SELECT unnest(string_split_regex(content, '\\r?\\n')) AS line
        FROM read_text('%(TMP_SURI_LOG)s')
        WHERE content IS NOT NULL AND content != ''
    ),
    parsed AS (
        SELECT
            COALESCE(
                TRY_STRPTIME(regexp_extract(line, '^([0-9]{2}/[0-9]{2}/[0-9]{4}-[0-9]{2}:[0-9]{2}:[0-9]{2})', 1), '%%m/%%d/%%Y-%%H:%%M:%%S'),
                date_trunc('second', CURRENT_TIMESTAMP)
            ) AS event_date,
            UPPER(COALESCE(NULLIF(regexp_extract(line, '\\[(Drop|Blocked|Allowed|Alert)\\]', 1), ''), 'ALERT')) AS action,
            TRY_CAST(regexp_extract(line, '\\[\\d+:(\\d+):\\d+\\]', 1) AS INTEGER) AS sid,
            regexp_extract(line, '\\[\\*\\*\\]\\s*\\[\\d+:\\d+:\\d+\\]\\s*(.*?)\\s*\\[\\*\\*\\]', 1) AS signature,
            regexp_extract(line, '\\[Classification:\\s*(.*?)\\]', 1) AS classification,
            TRY_CAST(regexp_extract(line, '\\[Priority:\\s*(\\d+)\\]', 1) AS INTEGER) AS priority,
            regexp_extract(line, '\\{([A-Za-z]+)\\}', 1) AS protocol,
            regexp_extract(line, '([0-9]{1,3}\\.[0-9]{1,3}\\.[0-9]{1,3}\\.[0-9]{1,3}):?\\d*\\s*->', 1) AS src_ip,
            TRY_CAST(regexp_extract(line, '([0-9]{1,3}\\.[0-9]{1,3}\\.[0-9]{1,3}\\.[0-9]{1,3}):(\\d+)\\s*->', 2) AS INTEGER) AS src_port,
            regexp_extract(line, '->\\s*([0-9]{1,3}\\.[0-9]{1,3}\\.[0-9]{1,3}\\.[0-9]{1,3})', 1) AS dst_ip,
            TRY_CAST(regexp_extract(line, '->\\s*[0-9]{1,3}\\.[0-9]{1,3}\\.[0-9]{1,3}\\.[0-9]{1,3}:(\\d+)', 1) AS INTEGER) AS dst_port
        FROM suri_lines
    )
    INSERT INTO log_suricata (
        event_date, action, sid, signature, classification, priority,
        protocol, src_ip, src_port, dst_ip, dst_port,
        src_country_code, src_country_name, dst_country_code, dst_country_name
    )
    SELECT DISTINCT
        p.event_date, p.action, p.sid, p.signature, p.classification, p.priority,
        p.protocol, p.src_ip, p.src_port, p.dst_ip, p.dst_port,
        'PENDING' AS src_country_code, 'Processing...' AS src_country_name,
        'PENDING' AS dst_country_code, 'Processing...' AS dst_country_name
    FROM parsed p
    WHERE p.event_date IS NOT NULL
      AND p.signature IS NOT NULL AND p.signature != ''
      AND (p.src_ip IS NOT NULL AND p.src_ip != '' OR p.dst_ip IS NOT NULL AND p.dst_ip != '')
      AND NOT EXISTS (
          SELECT 1 FROM log_suricata e
          WHERE e.event_date = p.event_date
            AND COALESCE(e.src_ip, '') = COALESCE(p.src_ip, '')
            AND COALESCE(e.dst_ip, '') = COALESCE(p.dst_ip, '')
            AND COALESCE(e.src_port, 0) = COALESCE(p.src_port, 0)
            AND COALESCE(e.dst_port, 0) = COALESCE(p.dst_port, 0)
            AND COALESCE(e.signature, '') = COALESCE(p.signature, '')
      );
    """ % {
        "TMP_SURI_LOG":   TMP_SURI_LOG,
        "DUCKDB_THREADS": DUCKDB_THREADS,
    }

    tmp_sql_fd, tmp_sql_path = tempfile.mkstemp(suffix=".sql")
    try:
        with os.fdopen(tmp_sql_fd, "w") as tmp_sql_file:
            tmp_sql_file.write(SQL_SURICATA)
        with open(tmp_sql_path, "r") as sql_input:
            proc = subprocess.Popen([DUCKDB_BIN, DB_SURICATA], stdin=sql_input, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
            _, stderr_out = proc.communicate()
            suri_status = proc.returncode
    finally:
        if os.path.exists(tmp_sql_path):
            os.remove(tmp_sql_path)
    if suri_status != 0:
        sys.stderr.write("[-] Suricata ingestion error:\n" + stderr_out.decode("utf-8", errors="replace"))
else:
    print("[+] No new Suricata logs to process.")

for f in (TMP_LOG, TMP_MSG_LOG, TMP_SURI_LOG):
    if os.path.exists(f):
        os.remove(f)

def run_geoip_enrichment(db_path, table_name):
    print("[+] Checking GeoIP cache status in %s..." % os.path.basename(db_path))
    must_reload = True
    try:
        proc_chk = subprocess.Popen([DUCKDB_BIN, db_path, "SELECT version FROM geoip_metadata;"], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
        stdout_chk, _ = proc_chk.communicate()
        if GEOIP_VERSION_TAG in stdout_chk.decode('utf-8'):
            proc_cnt = subprocess.Popen([DUCKDB_BIN, db_path, "SELECT COUNT(*) FROM geoip_cache;"], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
            stdout_cnt, _ = proc_cnt.communicate()
            if b"0" not in stdout_cnt:
                must_reload = False
    except Exception:
        pass

    SQL_GEOIP = "PRAGMA threads=%d;\n" % DUCKDB_THREADS
    SQL_GEOIP += "CREATE TABLE IF NOT EXISTS geoip_metadata (version VARCHAR);\n"
    SQL_GEOIP += "CREATE TABLE IF NOT EXISTS geoip_cache (ip_start BIGINT, ip_end BIGINT, country_code VARCHAR, country_name VARCHAR);\n"
    SQL_GEOIP += "UPDATE %(TABLE)s SET src_country_code = 'PENDING', src_country_name = 'Processing...' WHERE src_country_code = 'LAN' AND src_ip NOT SIMILAR TO '(10\\..*|192\\.168\\..*|172\\.(1[6-9]|2[0-9]|3[0-1])\\..*|127\\..*)';\n" % {"TABLE": table_name}
    SQL_GEOIP += "UPDATE %(TABLE)s SET dst_country_code = 'PENDING', dst_country_name = 'Processing...' WHERE dst_country_code = 'LAN' AND dst_ip NOT SIMILAR TO '(10\\..*|192\\.168\\..*|172\\.(1[6-9]|2[0-9]|3[0-1])\\..*|127\\..*)';\n" % {"TABLE": table_name}

    if must_reload:
        print("[!] Loading and indexing MaxMind GeoIP CSVs in %s..." % os.path.basename(db_path))
        SQL_GEOIP += "DROP TABLE IF EXISTS geoip_cache;\n"
        SQL_GEOIP += """
        CREATE TABLE geoip_cache AS 
        SELECT
            ( Prague.ip_parts[1]::BIGINT << 24 ) + ( Prague.ip_parts[2]::BIGINT << 16 ) + ( Prague.ip_parts[3]::BIGINT << 8 ) + Prague.ip_parts[4]::BIGINT AS ip_start,
            (( Prague.ip_parts[1]::BIGINT << 24 ) + ( Prague.ip_parts[2]::BIGINT << 16 ) + ( Prague.ip_parts[3]::BIGINT << 8 ) + Prague.ip_parts[4]::BIGINT) + (power(2, 32 - Prague.mask) - 1)::BIGINT AS ip_end,
            loc.country_iso_code AS country_code,
            loc.country_name     AS country_name
        FROM (
            SELECT 
                string_to_array(string_split(network, '/')[1], '.') AS ip_parts,
                string_split(network, '/')[2]::INTEGER AS mask,
                registered_country_geoname_id,
                geoname_id
            FROM read_csv('%(GEOIP_BLOCKS_V4)s', AUTO_DETECT=TRUE)
        ) AS Prague
        JOIN read_csv('%(GEOIP_LOCATIONS)s', AUTO_DETECT=TRUE) AS loc
          ON COALESCE(NULLIF(Prague.geoname_id::VARCHAR, ''), Prague.registered_country_geoname_id::VARCHAR) = loc.geoname_id::VARCHAR
        WHERE loc.country_iso_code IS NOT NULL AND loc.country_iso_code != ''
        ORDER BY ip_start, ip_end;\n""" % {"GEOIP_BLOCKS_V4": GEOIP_BLOCKS_V4, "GEOIP_LOCATIONS": GEOIP_LOCATIONS}
        SQL_GEOIP += "DELETE FROM geoip_metadata; INSERT INTO geoip_metadata VALUES ('%s');\n" % GEOIP_VERSION_TAG
    else:
        print("[+] GeoIP cache is up to date in %s." % os.path.basename(db_path))

    print("[+] Running fast IP lookup match (BIGINT) in %s..." % os.path.basename(db_path))
    SQL_GEOIP += """
    CREATE OR REPLACE TEMP TABLE pending_ips AS
    SELECT DISTINCT ip,
        (parts[1]::BIGINT << 24) + (parts[2]::BIGINT << 16) + (parts[3]::BIGINT << 8) + parts[4]::BIGINT AS ip_num
    FROM (
        SELECT ip, string_to_array(ip, '.') AS parts
        FROM (
            SELECT src_ip AS ip FROM %(TABLE)s WHERE src_country_code = 'PENDING'
            UNION
            SELECT dst_ip AS ip FROM %(TABLE)s WHERE dst_country_code = 'PENDING'
        )
        WHERE ip SIMILAR TO '[0-9]+\\.[0-9]+\\.[0-9]+\\.[0-9]+'
    ) AS parsed;

    CREATE OR REPLACE TEMP TABLE resolved_ips AS
    SELECT ip, country_code, country_name FROM (
        SELECT p.ip, geo.country_code, geo.country_name,
               ROW_NUMBER() OVER (PARTITION BY p.ip ORDER BY (geo.ip_end - geo.ip_start) ASC) AS rn
        FROM pending_ips AS p
        JOIN geoip_cache AS geo ON p.ip_num >= geo.ip_start AND p.ip_num <= geo.ip_end
    ) WHERE rn = 1;

    UPDATE %(TABLE)s AS target SET src_country_code = r.country_code, src_country_name = r.country_name FROM resolved_ips AS r WHERE target.src_country_code = 'PENDING' AND target.src_ip = r.ip;
    UPDATE %(TABLE)s AS target SET dst_country_code = r.country_code, dst_country_name = r.country_name FROM resolved_ips AS r WHERE target.dst_country_code = 'PENDING' AND target.dst_ip = r.ip;
    
    UPDATE %(TABLE)s SET src_country_code = 'LAN', src_country_name = 'Local Network' WHERE src_country_code = 'PENDING' AND src_ip IS NOT NULL AND src_ip != '';
    UPDATE %(TABLE)s SET dst_country_code = 'LAN', dst_country_name = 'Local Network' WHERE dst_country_code = 'PENDING' AND dst_ip IS NOT NULL AND dst_ip != '';
    
    UPDATE %(TABLE)s SET src_country_code = 'UNKNOWN', src_country_name = 'Unknown' WHERE src_country_code = 'PENDING';
    UPDATE %(TABLE)s SET dst_country_code = 'UNKNOWN', dst_country_name = 'Unknown' WHERE dst_country_code = 'PENDING';
    """ % {"TABLE": table_name}

    tmp_geo_fd, tmp_geo_path = tempfile.mkstemp(suffix=".sql")
    try:
        with os.fdopen(tmp_geo_fd, "w") as tmp_geo_file:
            tmp_geo_file.write(SQL_GEOIP)
        with open(tmp_geo_path, "r") as geo_input:
            proc_geo = subprocess.Popen([DUCKDB_BIN, db_path], stdin=geo_input, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
            _, stderr_geo = proc_geo.communicate()
            geo_status = proc_geo.returncode
    finally:
        if os.path.exists(tmp_geo_path):
            os.remove(tmp_geo_path)

    if geo_status != 0:
        sys.stderr.write("[-] GeoIP enrichment error in %s:\n" % os.path.basename(db_path) + stderr_geo.decode("utf-8", errors="replace"))
    else:
        print("[+] IP Geolocation enrichment completed successfully for %s." % os.path.basename(db_path))

if GEOIP_CSV_DIR:
    if fw_status == 0:
        run_geoip_enrichment(DB_FIREWALL, "log_firewall")
    if suri_status == 0:
        run_geoip_enrichment(DB_SURICATA, "log_suricata")

print("[+] Executing data retention cleanup policy...")
keep_days = get_logwatch_keep_days(default_days=30)
print("[+] Applied LOGWATCH_KEEP retention threshold: %d days" % keep_days)

if fw_status == 0 and os.path.exists(DB_FIREWALL):
    p_clean = subprocess.Popen([DUCKDB_BIN, DB_FIREWALL], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    sql_clean_fw = ("PRAGMA threads=%%d; DELETE FROM log_firewall WHERE event_date < CURRENT_DATE - INTERVAL %d DAY;" % keep_days) % DUCKDB_THREADS
    p_clean.communicate(input=sql_clean_fw.encode("utf-8"))

if msg_status == 0 and os.path.exists(DB_MESSAGES):
    p_clean = subprocess.Popen([DUCKDB_BIN, DB_MESSAGES], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    sql_clean_msg = ("PRAGMA threads=%%d; DELETE FROM log_messages WHERE event_date < CURRENT_DATE - INTERVAL %d DAY;" % keep_days) % DUCKDB_THREADS
    p_clean.communicate(input=sql_clean_msg.encode("utf-8"))

if suri_status == 0 and os.path.exists(DB_SURICATA):
    p_clean = subprocess.Popen([DUCKDB_BIN, DB_SURICATA], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    sql_clean_suri = ("PRAGMA threads=%%d; DELETE FROM log_suricata WHERE event_date < CURRENT_DATE - INTERVAL %d DAY;" % keep_days) % DUCKDB_THREADS
    p_clean.communicate(input=sql_clean_suri.encode("utf-8"))

import grp
import pwd
for db_f in (DB_FIREWALL, DB_MESSAGES, DB_SURICATA):
    if os.path.exists(db_f):
        try:
            uid = pwd.getpwnam("root").pw_uid
            gid = grp.getgrnam("nogroup").gr_gid
            os.chown(db_f, uid, gid)
        except KeyError:
            pass
        os.chmod(db_f, stat.S_IRUSR | stat.S_IWUSR | stat.S_IRGRP | stat.S_IWGRP | stat.S_IROTH)

release_lock()

end_time = time.time()
print("[+] ETL processing completed successfully in %.2f seconds!" % (end_time - start_time))
print("[+] Databases updated independently: firewall.db, messages.db, and suricata.db.")