feat: add WHOIS subnet querying and IP consolidation utility
- Implemented a new utility for querying WHOIS information and consolidating IP addresses by subnet. - Added caching mechanism for WHOIS responses to improve performance. - Introduced functions for loading and saving the WHOIS cache, querying subnets, and consolidating IPs based on a threshold. feat: create Flask web UI for intrusion detection system - Developed a Flask web application to manage the intrusion detection system. - Implemented routes for scanning, reports, blocklist management, and firewall configuration. - Added helper functions for reading files and handling requests. feat: define routes for web UI - Registered routes for various web pages and API endpoints in the Flask application. - Implemented functionality for scanning, reporting, blocklist management, and firewall state retrieval. feat: implement background scan runner for web UI - Created a background scan runner to handle scan jobs in a separate thread. - Managed an in-memory job store for tracking scan jobs and their statuses. - Added persistence for scan jobs to maintain state across application restarts. chore: update start script for Flask application - Enhanced the PowerShell start script to check for the existence of the virtual environment. - Added comments and structured the script for better readability and usability.
This commit is contained in:
1 parent
41230ce04c
commit
9f8ab0f61c
102 files changed
+1102652
-268
No files matched your search
@@ -0,0 +1,104 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Filesystem and I/O helpers for the intrusion-detection system."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import csv
|
||||
import gzip
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
from typing import Any, Iterable, Iterator, Sequence
|
||||
|
||||
ANSI_RE = re.compile(r"\x1b\[[0-9;]*m")
|
||||
|
||||
|
||||
# -- directory --------------------------------------------------------------
|
||||
|
||||
def validate_dir(path: str, need_write: bool = False) -> str:
|
||||
if not os.path.exists(path):
|
||||
raise FileNotFoundError(f"path does not exist: {path}")
|
||||
if not os.path.isdir(path):
|
||||
raise NotADirectoryError(f"path is not a directory: {path}")
|
||||
if not os.access(path, os.R_OK):
|
||||
raise PermissionError(f"path not readable: {path}")
|
||||
if need_write and not os.access(path, os.W_OK):
|
||||
raise PermissionError(f"path not writable: {path}")
|
||||
return path
|
||||
|
||||
|
||||
def ensure_dirs(*paths: str) -> None:
|
||||
for p in paths:
|
||||
os.makedirs(p, exist_ok=True)
|
||||
|
||||
|
||||
# -- log discovery ----------------------------------------------------------
|
||||
|
||||
def find_logs(path: str, pattern: re.Pattern | None = None,
|
||||
suffixes: Sequence[str] = (),
|
||||
exclude_prefixes: Sequence[str] = ()) -> list[str]:
|
||||
"""List matching log file names (not full paths), sorted."""
|
||||
out = []
|
||||
for name in os.listdir(path):
|
||||
if not os.path.isfile(os.path.join(path, name)):
|
||||
continue
|
||||
if any(name.startswith(p) for p in exclude_prefixes):
|
||||
continue
|
||||
if pattern is not None and not pattern.match(name):
|
||||
continue
|
||||
if suffixes and not name.endswith(tuple(suffixes)):
|
||||
continue
|
||||
out.append(name)
|
||||
return sorted(out)
|
||||
|
||||
|
||||
def read_lines(path: str) -> Iterator[str]:
|
||||
"""Yield ANSI-stripped, newline-stripped lines. Transparently gunzips."""
|
||||
opener = gzip.open if path.endswith(".gz") else open
|
||||
with opener(path, "rt", encoding="utf-8", errors="replace") as f:
|
||||
for line in f:
|
||||
yield ANSI_RE.sub("", line.rstrip("\r\n"))
|
||||
|
||||
|
||||
def read_lines_from(path: str, start: int = 0) -> Iterator[tuple[str, int]]:
|
||||
"""Yield ``(line, resume_offset)`` for a plain-text log from byte ``start``.
|
||||
|
||||
Only seekable (non-gzip) files can be resumed. ``resume_offset`` points
|
||||
after the last complete line; a trailing partial line (no newline yet) is
|
||||
yielded but resets the resume offset to its start so it is re-read once
|
||||
completed. The last yielded ``resume_offset`` is the value to store.
|
||||
"""
|
||||
with open(path, "rb") as f:
|
||||
f.seek(start)
|
||||
pos = start
|
||||
for raw in f:
|
||||
line_start = pos
|
||||
pos += len(raw)
|
||||
complete = raw.endswith(b"\n") or raw.endswith(b"\r")
|
||||
text = ANSI_RE.sub(
|
||||
"", raw.decode("utf-8", errors="replace").rstrip("\r\n"))
|
||||
yield text, (pos if complete else line_start)
|
||||
|
||||
|
||||
# -- writers ----------------------------------------------------------------
|
||||
|
||||
def write_csv(path: str, header: Sequence[str],
|
||||
rows: Iterable[Sequence[Any]]) -> str:
|
||||
with open(path, "w", newline="", encoding="utf-8") as f:
|
||||
w = csv.writer(f)
|
||||
w.writerow(header)
|
||||
w.writerows(rows)
|
||||
return path
|
||||
|
||||
|
||||
def write_json(path: str, obj: Any) -> str:
|
||||
with open(path, "w", encoding="utf-8") as f:
|
||||
json.dump(obj, f, indent=2, default=str)
|
||||
return path
|
||||
|
||||
|
||||
def write_lines(path: str, lines: Iterable[str]) -> str:
|
||||
with open(path, "w", encoding="utf-8") as f:
|
||||
for line in lines:
|
||||
f.write(line + "\n")
|
||||
return path
|
||||
Reference in new issue
Block a user