Advanced Application Development

Transactions

Трансакциите во системот се имплементирани на апликациско ниво преку SQLite конекции. Секоја операција која вклучува повеќе табели се извршува атомарно — доколку некоја од операциите не успее, се повикува conn.rollback() и ниедна промена не се зачувува.

---

Прием на податоци од агент (/receive)

Најкритичната трансакција во системот се извршува при прием на податоци од агентите преку POST /receive. Во рамките на една конекција се извршуваат следните операции:

• Пронаоѓање или креирање на компјутер во computers
• Ажурирање на last_seen, ip, os, user, sysmon_available
• Внесување на системски метрики во computer_history
• Бришење и повторно внесување на тековните процеси во computer_processes_current
• Условно внесување на историски процеси во computer_processes_history (доколку е овозможено во env_settings)
• Внесување на Sysmon настани во sysmon_events
• Внесување на мрежни конекции во network_connections

Сите операции се завршуваат со conn.commit(). Доколку се случи грешка во кој било чекор, се извршува conn.rollback() и ниедна промена не е трајно зачувана:

@app.route("/receive", methods=["POST"])
def receive_data():
    conn = None
    try:
        data = request.get_json(force=True, silent=True) or {}
        env_name, tenant_id = get_env_from_token(request)

        conn = db()
        c = conn.cursor()

        # Чекор 1: Пронаоѓање или креирање на компјутер
        c.execute(
            "SELECT id FROM computers WHERE tenant_id=? AND name=? LIMIT 1",
            (tenant_id, computer_name),
        )
        row = c.fetchone()

        if row:
            computer_id = row["id"]
            c.execute("""
                UPDATE computers
                SET user=?, ip=?, os=?, last_seen=?, sysmon_available=?, env_name=?
                WHERE id=? AND tenant_id=?
            """, (info.get("user"), info.get("ip_address"), info.get("os"),
                  now_iso, int(info.get("is_sysmon_available", 0) or 0),
                  env_name, computer_id, tenant_id))
        else:
            c.execute("""
                INSERT INTO computers(tenant_id, env_name, name, user, ip, os,
                                      first_seen, last_seen, sysmon_available)
                VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?)
            """, (tenant_id, env_name, computer_name, info.get("user"),
                  info.get("ip_address"), info.get("os"), now_iso, now_iso,
                  int(info.get("is_sysmon_available", 0) or 0)))
            computer_id = c.lastrowid

        # Чекор 2: Системски метрики
        c.execute("""
            INSERT INTO computer_history(
                computer_id, cpu_usage, ram_usage, disk_usage,
                network_sent_mb, network_recv_mb, timestamp)
            VALUES(?, ?, ?, ?, ?, ?, ?)
        """, (computer_id, float(info.get("cpu_usage") or 0),
              float(info.get("ram_usage") or 0), float(info.get("disk_usage") or 0),
              float(info.get("network_sent_mb") or 0),
              float(info.get("network_recv_mb") or 0),
              info.get("timestamp") or now_iso))

        # Чекор 3: Тековни процеси (бриши и внеси повторно)
        c.execute("DELETE FROM computer_processes_current WHERE computer_id=?", (computer_id,))
        for proc in (data.get("processes") or []):
            c.execute("""
                INSERT INTO computer_processes_current(
                    computer_id, pid, name, cpu_percent,
                    memory_mb, username, cmdline, timestamp)
                VALUES(?, ?, ?, ?, ?, ?, ?, ?)
            """, (computer_id, proc.get("pid"), proc.get("name"),
                  float(proc.get("cpu_percent") or 0),
                  float(proc.get("memory_mb") or 0),
                  proc.get("user") or proc.get("username"),
                  proc.get("cmdline"), info.get("timestamp") or now_iso))

        # Чекор 4: Историски процеси (само ако е овозможено во env_settings)
        if is_process_history_enabled(tenant_id, env_name):
            for proc in (data.get("processes") or []):
                c.execute("""
                    INSERT INTO computer_processes_history(
                        computer_id, pid, name, cpu_percent,
                        memory_mb, username, cmdline, timestamp)
                    VALUES(?, ?, ?, ?, ?, ?, ?, ?)
                """, (computer_id, proc.get("pid"), proc.get("name"),
                      float(proc.get("cpu_percent") or 0),
                      float(proc.get("memory_mb") or 0),
                      proc.get("user") or proc.get("username"),
                      proc.get("cmdline"), info.get("timestamp") or now_iso))

        # Чекор 5: Sysmon настани
        for ev in (security_data.get("sysmon_events") or []):
            c.execute("""
                INSERT INTO sysmon_events(
                    computer_id, event_id, event_type, message, timestamp, details)
                VALUES(?, ?, ?, ?, ?, ?)
            """, (computer_id, ev.get("event_id"), ev.get("event_type", "Unknown"),
                  ev.get("message", ""),
                  ev.get("timestamp") or (info.get("timestamp") or now_iso),
                  json.dumps(ev, ensure_ascii=False)))

        # Чекор 6: Мрежни конекции
        for nc in (security_data.get("network_connections") or []):
            c.execute("""
                INSERT INTO network_connections(
                    computer_id, pid, local_address,
                    remote_address, status, process_name, timestamp)
                VALUES(?, ?, ?, ?, ?, ?, ?)
            """, (computer_id, nc.get("pid"), nc.get("local_address"),
                  nc.get("remote_address"), nc.get("status"),
                  nc.get("process_name"), info.get("timestamp") or now_iso))

        # Сите операции успешни — зачувај
        conn.commit()
        return jsonify({"ok": True, "computer_id": computer_id, ...})

    except Exception as e:
        if conn:
            conn.rollback()   # Врати ги сите промени
        return jsonify({"error": str(e)}), 500

    finally:
        if conn:
            conn.close()

Клучни карактеристики на оваа трансакција:

• conn.rollback() се повикува во except блокот — ниедна табела нема делумни податоци
• conn.close() е во finally блокот — конекцијата секогаш се затвора
• Условното снимање на процесна историја (is_process_history_enabled) е дел од истата трансакција

---

Google автентикација и автоматско креирање на tenant (/api/auth/google)

При прва најава на нов корисник, во рамките на една конекција атомарно се извршуваат:

• Внесување или ажурирање на корисникот во users
• Проверка за постојно членство во memberships
• Доколку нов корисник: креирање на tenant, membership и default environment

@app.route("/api/auth/google", methods=["POST"])
def auth_google():
    try:
        idinfo = id_token.verify_oauth2_token(credential, grequests.Request(), GOOGLE_CLIENT_ID)
        email = idinfo.get("email")

        conn = db()
        c = conn.cursor()

        # Чекор 1: Upsert корисник
        c.execute("SELECT id FROM users WHERE email=?", (email,))
        row = c.fetchone()
        if row:
            user_id = row["id"]
            c.execute("UPDATE users SET name=?, picture=? WHERE id=?", (name, picture, user_id))
        else:
            c.execute(
                "INSERT INTO users(email, name, picture, created_at) VALUES(?,?,?,datetime('now'))",
                (email, name, picture))
            user_id = c.lastrowid

        # Чекор 2: Проверка за membership
        c.execute("SELECT tenant_id, role FROM memberships WHERE user_id=? LIMIT 1", (user_id,))
        m = c.fetchone()

        if not m:
            # Чекор 3: Нов корисник — креирај tenant, membership и default environment атомарно
            c.execute(
                "INSERT INTO tenants(name, owner_email, created_at) VALUES(?,?,datetime('now'))",
                (tenant_name, email))
            tenant_id = c.lastrowid

            c.execute(
                "INSERT INTO memberships(user_id, tenant_id, role, created_at) VALUES(?,?,'admin',datetime('now'))",
                (user_id, tenant_id))

            c.execute(
                "INSERT OR IGNORE INTO environments(tenant_id, name, created_at) VALUES(?,'default',datetime('now'))",
                (tenant_id,))
        else:
            tenant_id = m["tenant_id"]
            role = m["role"] or "admin"

        conn.commit()
        conn.close()

        session = make_jwt({"uid": user_id, "email": email, "role": role, "tenant_id": tenant_id})
        ...

    except Exception as e:
        return jsonify({"error": "Invalid Google token", "details": str(e)}), 401

Доколку некоја од INSERT операциите не успее, целата автентикација се откажува и корисникот добива грешка.

---

Генерирање на environment токен (/api/admin/tokens)

При генерирање на нов токен за агент, системот прво верификува дека environment постои, а потоа внесува токен. Токенот се генерира со secrets.token_urlsafe(32) и е валиден 90 дена:

@app.route("/api/admin/tokens", methods=["POST"])
@require_tenant_admin()
def admin_generate_token():
    tenant_id = request.user["tenant_id"]
    env = (data.get("env") or "").strip()

    conn = db()
    c = conn.cursor()

    # Верификација дека environment постои
    c.execute("SELECT 1 FROM environments WHERE tenant_id=? AND name=? LIMIT 1", (tenant_id, env))
    if not c.fetchone():
        conn.close()
        return jsonify({"error": "Environment not found"}), 404

    # Генерирање и зачувување на токенот
    token = secrets.token_urlsafe(32)
    c.execute("""
        INSERT INTO env_tokens(tenant_id, env_name, token, created_at, expires_at)
        VALUES(?, ?, ?, datetime('now'), datetime('now', '+90 days'))
    """, (tenant_id, env, token))
    conn.commit()
    conn.close()

    return jsonify({"ok": True, "env": env, "token": token})

Токенот се проверува при секој повик на агентот преку get_env_from_token():

def get_env_from_token(req):
    tok = req.headers.get("X-Env-Token", "")
    if not tok:
        return (None, None)

    conn = db()
    c = conn.cursor()
    c.execute("""
        SELECT env_name, tenant_id
        FROM env_tokens
        WHERE token = ?
          AND (expires_at IS NULL OR expires_at > datetime('now'))
        LIMIT 1
    """, (tok,))
    row = c.fetchone()
    conn.close()

    if not row:
        return (None, None)
    return (row["env_name"], row["tenant_id"])

---

Ажурирање на поставки за околина (/api/admin/env-settings/<env_name>)

Поставките за околина се запишуваат атомарно преку INSERT OR ... ON CONFLICT DO UPDATE (upsert), со цел да се избегне race condition при истовремено читање и пишување:

@app.route("/api/admin/env-settings/<env_name>", methods=["POST"])
@require_tenant_admin()
def admin_set_env_settings(env_name):
    tenant_id = request.user["tenant_id"]
    enabled = 1 if bool(data.get("save_process_history")) else 0

    conn = db()
    c = conn.cursor()
    c.execute("""
        INSERT INTO env_settings(tenant_id, env_name, save_process_history, created_at, updated_at)
        VALUES(?, ?, ?, datetime('now'), datetime('now'))
        ON CONFLICT(tenant_id, env_name) DO UPDATE SET
            save_process_history=excluded.save_process_history,
            updated_at=datetime('now')
    """, (tenant_id, env_name, enabled))
    conn.commit()
    conn.close()

    return jsonify({"ok": True, "env": env_name, "save_process_history": bool(enabled)})

---

Database Connection Pooling

Во моменталната имплементација системот користи SQLite база на податоци преку sqlite3 библиотеката. Конекцијата кон базата се воспоставува преку помошната функција db():

DB_FILE = os.environ.get("DB_FILE", "lan_logs_sysmon.db")

def db():
    conn = sqlite3.connect(DB_FILE)
    conn.row_factory = sqlite3.Row
    return conn

conn.row_factory = sqlite3.Row овозможува пристап до колоните по име (на пр. row["email"]) наместо по индекс. При секое барање се отвора нова конекција која се затвора по завршување на операцијата — секогаш во finally блокот:

conn = db()
try:
    c = conn.cursor()
    c.execute("SELECT ...")
    result = c.fetchall()
    conn.commit()
finally:
    conn.close()

Поради природата на SQLite (file-based база со writer lock), Connection Pooling не е применливо во тековната верзија.

---

Позадинска задача за автоматско чистење (cleanup_old_data)

Системот стартува посебна daemon нишка која на секои 3600 секунди (1 час) автоматски ги брише записите постари од 30 дена. Нишката ја користи истата db() функција и работи со своја независна конекција:

def cleanup_old_data():
    """Deletes old rows (30 days) every hour."""
    while True:
        time.sleep(3600)
        try:
            conn = db()
            c = conn.cursor()
            cutoff = (datetime.now() - timedelta(days=30)).isoformat()

            c.execute("DELETE FROM computer_history WHERE timestamp < ?", (cutoff,))
            c.execute("DELETE FROM sysmon_events WHERE timestamp < ?", (cutoff,))
            c.execute("DELETE FROM network_connections WHERE timestamp < ?", (cutoff,))
            c.execute("DELETE FROM computer_processes_history WHERE timestamp < ?", (cutoff,))

            conn.commit()
            conn.close()
            print(f"[Cleanup] Deleted entries older than {cutoff}")
        except Exception as e:
            print("[Cleanup] Error:", e)

# Стартување при иницијализација на серверот
cleanup_thread = threading.Thread(target=cleanup_old_data, daemon=True)
cleanup_thread.start()

Нишката е daemon=True — автоматски се гасне кога се гасне главниот Flask процес.

---

Планирана миграција кон PostgreSQL со Connection Pooling

По миграција кон PostgreSQL, db() функцијата би се заменила со SQLAlchemy engine со connection pool:

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from contextlib import contextmanager

DATABASE_URL = "postgresql://app_user:password@localhost:5432/netintel_db"

engine = create_engine(
    DATABASE_URL,
    pool_size=10,        # Број на постојни конекции во pool-от
    max_overflow=20,     # Дополнителни конекции при врв на оптоварување
    pool_timeout=30,     # Максимално чекање за слободна конекција (секунди)
    pool_recycle=1800,   # Освежување на конекции на секои 30 минути
    pool_pre_ping=True   # Проверка дали конекцијата е активна пред употреба
)

SessionLocal = sessionmaker(bind=engine)

@contextmanager
def db():
    session = SessionLocal()
    try:
        yield session
        session.commit()
    except Exception:
        session.rollback()
        raise
    finally:
        session.close()  # Конекцијата се враќа во pool-от, не се затвора физички

Пример на употреба во /receive endpoint со новиот pool:

@app.route("/receive", methods=["POST"])
def receive_data():
    with db() as session:
        result = session.execute(
            text("SELECT id FROM computers WHERE tenant_id=:tid AND name=:name LIMIT 1"),
            {"tid": tenant_id, "name": computer_name}
        ).fetchone()
        # Конекцијата автоматски се враќа во pool-от по излегување од with блокот

Предности на оваа миграција:

• Повторно користење на активни конекции наместо отворање нова при секое барање
• Намалување на латенцијата — воспоставувањето на нова PostgreSQL конекција трае ~50-100ms
• pool_pre_ping=True автоматски ги отстранува прекинатите конекции од pool-от
• Подобри перформанси при голем број истовремени агенти кои праќаат податоци паралелно
• threaded=True е веќе поставено на Flask серверот (app.run(..., threaded=True)), па pooling директно го подобрува паралелизмот