94 lines
3.2 KiB
Python
94 lines
3.2 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import sqlite3
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
from .models import Host
|
|
|
|
|
|
def data_dir() -> Path:
|
|
root = os.environ.get("XDG_DATA_HOME")
|
|
if root:
|
|
path = Path(root) / "librenet-scanner"
|
|
else:
|
|
path = Path.home() / ".local" / "share" / "librenet-scanner"
|
|
path.mkdir(parents=True, exist_ok=True)
|
|
return path
|
|
|
|
|
|
class HistoryStore:
|
|
def __init__(self, db_path: Path | None = None) -> None:
|
|
self.db_path = db_path or data_dir() / "history.sqlite3"
|
|
self.db_path.parent.mkdir(parents=True, exist_ok=True)
|
|
self._init_db()
|
|
|
|
def _connect(self) -> sqlite3.Connection:
|
|
conn = sqlite3.connect(self.db_path)
|
|
conn.row_factory = sqlite3.Row
|
|
return conn
|
|
|
|
def _init_db(self) -> None:
|
|
with self._connect() as conn:
|
|
conn.executescript(
|
|
"""
|
|
PRAGMA journal_mode=WAL;
|
|
CREATE TABLE IF NOT EXISTS scans (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
created_at TEXT NOT NULL,
|
|
target TEXT NOT NULL,
|
|
profile TEXT NOT NULL,
|
|
host_count INTEGER NOT NULL
|
|
);
|
|
CREATE TABLE IF NOT EXISTS scan_hosts (
|
|
scan_id INTEGER NOT NULL,
|
|
ip TEXT NOT NULL,
|
|
hostname TEXT,
|
|
mac TEXT,
|
|
vendor TEXT,
|
|
os_name TEXT,
|
|
ports_json TEXT NOT NULL,
|
|
FOREIGN KEY(scan_id) REFERENCES scans(id) ON DELETE CASCADE
|
|
);
|
|
"""
|
|
)
|
|
|
|
def save_scan(self, target: str, profile: str, hosts: list[Host]) -> int:
|
|
now = datetime.now(timezone.utc).astimezone().isoformat(timespec="seconds")
|
|
with self._connect() as conn:
|
|
cur = conn.execute(
|
|
"INSERT INTO scans(created_at, target, profile, host_count) VALUES (?, ?, ?, ?)",
|
|
(now, target, profile, len(hosts)),
|
|
)
|
|
scan_id = int(cur.lastrowid)
|
|
for host in hosts:
|
|
ports = [
|
|
{
|
|
"port": p.port,
|
|
"protocol": p.protocol,
|
|
"service": p.service,
|
|
"product": p.product,
|
|
"version": p.version,
|
|
}
|
|
for p in host.ports
|
|
]
|
|
conn.execute(
|
|
"""
|
|
INSERT INTO scan_hosts(scan_id, ip, hostname, mac, vendor, os_name, ports_json)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?)
|
|
""",
|
|
(scan_id, host.ip, host.hostname, host.mac, host.vendor, host.os_name, json.dumps(ports, ensure_ascii=False)),
|
|
)
|
|
return scan_id
|
|
|
|
def recent_scans(self, limit: int = 100) -> list[sqlite3.Row]:
|
|
with self._connect() as conn:
|
|
return list(
|
|
conn.execute(
|
|
"SELECT id, created_at, target, profile, host_count FROM scans ORDER BY id DESC LIMIT ?",
|
|
(limit,),
|
|
)
|
|
)
|