Changes between Initial Version and Version 1 of AdvancedApplicationDevelopment

Timestamp:

07/07/26 18:51:44 (6 hours ago)

Author:

231118

Comment:

--

AdvancedApplicationDevelopment

@@ +1 @@
+= 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 директно го подобрува паралелизмот