3 Commits
Author SHA1 Message Date
me0nline 082b839965 Configure separate Windows and Linux Gitea release workflows with dedicated SHA-256 checksums
CI Test Suite / Run Component Tests & Pipeline Verification (push) Successful in 3m48s
2026-09-04 15:46:35 +02:00
me0nline e7bf277fb9 Add v1.0.1 release notes documenting removal of client filter logic
CI Test Suite / Run Component Tests & Pipeline Verification (push) Successful in 1m36s
Release Binaries / release (push) Successful in 2m51s
2026-09-04 15:38:10 +02:00
me0nline 7052e68589 Implement edge filtering, state tracking, clean out/ directory, and add Gitea CI workflow
CI Test Suite / Run Component Tests & Pipeline Verification (push) Successful in 1m40s
2026-09-04 15:32:39 +02:00
30 changed files with 1660 additions and 1532 deletions
+73
View File
@@ -0,0 +1,73 @@
name: CI Test Suite
on:
push:
branches:
- '**'
tags-ignore:
- 'v*'
pull_request:
workflow_dispatch:
jobs:
test:
name: Run Component Tests & Pipeline Verification
runs-on: ubuntu-latest
steps:
- name: Checkout Code
uses: actions/checkout@v4
- name: Install System Dependencies & Python
run: |
if command -v apt-get >/dev/null 2>&1; then
apt-get update -y
apt-get install -y python3 python3-pip python3-venv curl
fi
python3 -m pip install --upgrade pip --break-system-packages || python3 -m pip install --upgrade pip || true
pip3 install -r requirements.txt --break-system-packages || pip3 install -r requirements.txt
- name: Verify Python Syntax
run: |
python3 -m py_compile Server.py Win_Client.py Linux_Client.py package_dist.py upload_release.py test_pipeline.py tests/*.py
- name: Run Component Unit Tests
run: |
python3 -m unittest discover -s tests -v
- name: Run End-to-End Pipeline Integration Test
run: |
# Clean up any leftover test configs or database
rm -f server_config.json client_config.json logar_state.db client_state.json
# 1. Initialize server config and export client configuration
python3 Server.py --create-client-config --server-host 127.0.0.1 --server-port 9443 --client-out client_config.json
# 2. Launch LOGAR server in the background
python3 Server.py &
SERVER_PID=$!
echo "[*] Server launched in background with PID $SERVER_PID"
# 3. Poll Hermes health / report endpoint until server is listening
READY=0
for i in $(seq 1 20); do
if curl -s http://127.0.0.1:8443/api/hermes/report >/dev/null 2>&1; then
echo "[+] LOGAR Server is ready after ${i}s."
READY=1
break
fi
sleep 1
done
if [ $READY -ne 1 ]; then
echo "[!] Server failed to start within 20 seconds."
kill $SERVER_PID || true
exit 1
fi
# 4. Execute end-to-end integration test
python3 test_pipeline.py
# 5. Cleanly terminate background server
kill $SERVER_PID || true
wait $SERVER_PID 2>/dev/null || true
echo "[+] Server stopped successfully."
@@ -1,19 +1,25 @@
name: Release Binaries
name: Release Linux Binaries
on:
push:
tags:
- 'v*'
workflow_dispatch:
inputs:
tag:
description: 'Release tag (e.g. v1.0.1)'
required: false
default: 'v1.0.1'
jobs:
release:
release-linux:
name: Build & Release Linux Binaries
runs-on: ubuntu-latest
steps:
- name: Checkout Code
uses: actions/checkout@v4
- name: Install Python and Dependencies
- name: Install Python and Build Dependencies
run: |
if command -v apt-get >/dev/null 2>&1; then
apt-get update -y
@@ -22,15 +28,15 @@ jobs:
python3 -m pip install --upgrade pip --break-system-packages || python3 -m pip install --upgrade pip || true
pip3 install pyinstaller -r requirements.txt --break-system-packages || pip3 install pyinstaller -r requirements.txt
- name: Compile Standalone Binaries
- name: Compile Standalone Linux Binaries
run: |
python3 package_dist.py
python3 package_dist.py --target linux
- name: Publish Release
- name: Publish Linux Release Assets
env:
GITEA_TOKEN: ${{ secrets.TAG_TOKEN || github.token }}
GITEA_SERVER_URL: ${{ github.server_url }}
GITEA_REPOSITORY: ${{ github.repository }}
GITEA_REF_NAME: ${{ github.ref_name }}
GITEA_REF_NAME: ${{ inputs.tag || github.ref_name }}
run: |
python3 upload_release.py --skip-build
+65
View File
@@ -0,0 +1,65 @@
name: Release Windows Binaries
on:
push:
tags:
- 'v*'
workflow_dispatch:
inputs:
tag:
description: 'Release tag (e.g. v1.0.1)'
required: false
default: 'v1.0.1'
jobs:
release-windows:
name: Build & Release Windows Binaries
runs-on: windows-latest
steps:
- name: Checkout Code
uses: actions/checkout@v4
- name: Set up Python
uses: actions/setup-python@v5
with:
python-version: '3.12'
continue-on-error: true
- name: Install Dependencies
shell: powershell
run: |
$py = "python"
if (-not (Get-Command "python" -ErrorAction SilentlyContinue)) {
if (Get-Command "py" -ErrorAction SilentlyContinue) {
$py = "py -3.12"
}
}
& $py -m pip install --upgrade pip
& $py -m pip install pyinstaller -r requirements.txt
- name: Compile Standalone Windows Binaries
shell: powershell
run: |
$py = "python"
if (-not (Get-Command "python" -ErrorAction SilentlyContinue)) {
if (Get-Command "py" -ErrorAction SilentlyContinue) {
$py = "py -3.12"
}
}
& $py package_dist.py --target windows
- name: Publish Windows Release Assets
shell: powershell
env:
GITEA_TOKEN: ${{ secrets.TAG_TOKEN || github.token }}
GITEA_SERVER_URL: ${{ github.server_url }}
GITEA_REPOSITORY: ${{ github.repository }}
GITEA_REF_NAME: ${{ inputs.tag || github.ref_name }}
run: |
$py = "python"
if (-not (Get-Command "python" -ErrorAction SilentlyContinue)) {
if (Get-Command "py" -ErrorAction SilentlyContinue) {
$py = "py -3.12"
}
}
& $py upload_release.py --skip-build
+2
View File
@@ -20,6 +20,8 @@ dist/
*.db
*.sqlite
*.sqlite3
client_state.json
*.tmp
# Live configuration with generated private keys & tokens (samples are tracked)
server_config.json
+155 -18
View File
@@ -6,7 +6,8 @@ import struct
import argparse
import subprocess
import warnings
from datetime import datetime, timezone
from datetime import datetime, timezone, timedelta
from typing import Optional, Dict, Any, List
# Suppress cryptography / pgpy deprecation notices
warnings.filterwarnings("ignore")
@@ -14,6 +15,49 @@ warnings.filterwarnings("ignore")
import pgpy
CONFIG_FILE_NAME = "client_config.json"
STATE_FILE_NAME = "client_state.json"
def get_state_path(config_path: str, custom_state_path: Optional[str] = None) -> str:
if custom_state_path:
return custom_state_path
config_dir = os.path.dirname(os.path.abspath(config_path))
return os.path.join(config_dir, STATE_FILE_NAME)
def load_state(state_path: str) -> dict:
if os.path.exists(state_path):
try:
with open(state_path, "r", encoding="utf-8") as f:
return json.load(f)
except Exception as e:
print(f"[!] Warning: Failed to read state file '{state_path}': {e}")
return {}
return {}
def save_state(state_path: str, state: dict):
try:
temp_path = f"{state_path}.tmp"
with open(temp_path, "w", encoding="utf-8") as f:
json.dump(state, f, indent=2)
os.replace(temp_path, state_path)
except Exception as e:
print(f"[!] Warning: Could not save client state to '{state_path}': {e}")
def commit_state(state: dict, state_path: str):
if "new_last_cursor" in state:
val = state.pop("new_last_cursor")
if val:
state["last_cursor"] = val
if "new_last_timestamp_us" in state:
val = state.pop("new_last_timestamp_us")
if val:
state["last_timestamp_us"] = val
if "new_sent_cursors" in state:
state["sent_cursors"] = state.pop("new_sent_cursors")
save_state(state_path, state)
def load_config(config_path: str = CONFIG_FILE_NAME):
@@ -63,12 +107,40 @@ def get_machine_identifier() -> str:
return hostname
def get_recent_linux_logs(hours: int = 6):
def get_recent_linux_logs(hours: int = 24, state: Optional[dict] = None) -> list:
"""
Collects warnings and errors from systemd journalctl over the lookback window.
Edge Thinness: Drops INFO and DEBUG entries at the source.
Collects info, warnings, and errors from systemd journalctl over the lookback window.
Edge Filtering: Retains INFO, WARNING, and ERROR. Strips DEBUG (priority 7) and skips events older than lookback window (default: 24h).
State Tracking: Skips events older than lookback window (default 24h) and events
that have already been sent in previous runs.
"""
cmd = ["journalctl", "--since", f"{hours} hours ago", "-p", "warning", "--output=json"]
last_cursor = None
last_timestamp_us = 0.0
sent_cursors = set()
if state:
last_cursor = state.get("last_cursor")
try:
last_timestamp_us = float(state.get("last_timestamp_us", 0))
except (ValueError, TypeError):
last_timestamp_us = 0.0
sent_cursors = set(state.get("sent_cursors", []))
cmd = ["journalctl", "--since", f"{hours} hours ago", "-p", "info", "--output=json"]
result = None
if last_cursor:
cmd_with_cursor = ["journalctl", "--since", f"{hours} hours ago", "--after-cursor", str(last_cursor), "-p", "info", "--output=json"]
try:
res = subprocess.run(cmd_with_cursor, capture_output=True, text=True, check=False)
if res.returncode == 0:
result = res
except FileNotFoundError:
print("[!] journalctl command not found. Ensure this script runs on a systemd-enabled Linux system.")
return []
except Exception:
pass
if result is None:
try:
result = subprocess.run(cmd, capture_output=True, text=True, check=False)
except FileNotFoundError:
@@ -80,6 +152,11 @@ def get_recent_linux_logs(hours: int = 6):
logs = []
machine_id = get_machine_identifier()
cutoff_epoch_us = (datetime.now(timezone.utc) - timedelta(hours=hours)).timestamp() * 1_000_000
newest_cursor = None
newest_timestamp_us = last_timestamp_us
collected_cursors = []
for line in result.stdout.splitlines():
line_str = line.strip()
@@ -87,13 +164,48 @@ def get_recent_linux_logs(hours: int = 6):
continue
try:
entry = json.loads(line_str)
priority = str(entry.get("PRIORITY", "4"))
# Priority 0: Emerg, 1: Alert, 2: Crit, 3: Err, 4: Warning.
# Strip anything above 4 (5: Notice, 6: Info, 7: Debug)
if int(priority) > 4:
entry_cursor = entry.get("__CURSOR")
entry_ts_us_raw = entry.get("__REALTIME_TIMESTAMP")
entry_ts_us = 0.0
if entry_ts_us_raw:
try:
entry_ts_us = float(entry_ts_us_raw)
except (ValueError, TypeError):
pass
# 1. Skip entries older than lookback window (default: 24h)
if entry_ts_us and entry_ts_us < cutoff_epoch_us:
continue
sev = "WARNING" if priority == "4" else "ERROR"
# 2. Skip already sent events
if entry_cursor and (entry_cursor in sent_cursors or entry_cursor == last_cursor):
continue
if last_timestamp_us > 0 and entry_ts_us > 0 and entry_ts_us < last_timestamp_us:
continue
# Advance newest tracking for new entries
if entry_cursor:
newest_cursor = entry_cursor
collected_cursors.append(entry_cursor)
if entry_ts_us > newest_timestamp_us:
newest_timestamp_us = entry_ts_us
priority = int(entry.get("PRIORITY", "6"))
# Priority 0: Emerg, 1: Alert, 2: Crit, 3: Err (-> ERROR)
# Priority 4: Warning, 5: Notice (-> WARNING)
# Priority 6: Info (-> INFO)
# Priority 7: Debug (skip)
if priority > 6:
continue
if priority <= 3:
sev = "ERROR"
elif priority in (4, 5):
sev = "WARNING"
else:
sev = "INFO"
logs.append({
"server": machine_id,
"os_type": "linux",
@@ -104,13 +216,19 @@ def get_recent_linux_logs(hours: int = 6):
except (json.JSONDecodeError, ValueError):
continue
if state is not None:
state["new_last_cursor"] = newest_cursor or last_cursor
state["new_last_timestamp_us"] = max(newest_timestamp_us, last_timestamp_us)
state["new_sent_cursors"] = (list(sent_cursors) + collected_cursors)[-1000:]
state["last_run_timestamp"] = datetime.now(timezone.utc).isoformat()
return logs
def send_encrypted_logs_over_socket(config: dict, logs: list):
"""
Encrypts the payload using the server's OpenPGP public key and streams
over an authenticated TCP socket. Zero local state is maintained on the client.
over an authenticated TCP socket.
"""
server_host = config["server_host"]
server_port = int(config["server_port"])
@@ -128,7 +246,7 @@ def send_encrypted_logs_over_socket(config: dict, logs: list):
machine_id = get_machine_identifier()
# Prepare zero-state candidate batch
# Prepare batch
payload = {
"server": machine_id,
"timestamp": datetime.now(timezone.utc).isoformat(),
@@ -178,9 +296,11 @@ def send_encrypted_logs_over_socket(config: dict, logs: list):
def main():
parser = argparse.ArgumentParser(description="LOGAR Linux Edge Log Forwarder (Zero State)")
parser = argparse.ArgumentParser(description="LOGAR Linux Edge Log Forwarder with State Tracking")
parser.add_argument("--config", default=CONFIG_FILE_NAME, help="Path to client_config.json")
parser.add_argument("--hours", type=int, default=6, help="Lookback window in hours for journalctl logs")
parser.add_argument("--hours", type=int, default=24, help="Lookback window in hours for journalctl logs (default: 24)")
parser.add_argument("--state-file", default=None, help="Path to state tracking file (default: client_state.json next to config)")
parser.add_argument("--no-state", action="store_true", help="Disable state tracking and send all events matching lookback window")
args = parser.parse_args()
try:
@@ -189,14 +309,31 @@ def main():
print(f"[!] Configuration error: {e}")
sys.exit(1)
state_path = get_state_path(args.config, args.state_file)
state = None if args.no_state else load_state(state_path)
machine_id = get_machine_identifier()
print(f"[*] Edge Forwarder Node: {machine_id}")
print(f"[*] Scanning Linux journalctl for candidate anomalies (last {args.hours} hours)...")
candidate_logs = get_recent_linux_logs(hours=args.hours)
print(f"[*] Found {len(candidate_logs)} candidate anomalies (noise stripped at source).")
if state and ("last_cursor" in state or "last_timestamp_us" in state):
print(f"[*] State tracking active: resuming after previous cursor/timestamp (state file: {state_path})")
elif not args.no_state:
print(f"[*] State tracking initialized (state file: {state_path})")
print(f"[*] Scanning Linux journalctl for unsent entries (last {args.hours} hours)...")
candidate_logs = get_recent_linux_logs(hours=args.hours, state=state)
print(f"[*] Found {len(candidate_logs)} unsent candidate entries (INFO to ERROR, entries > {args.hours}h and already-sent skipped).")
if not candidate_logs:
print("[*] No new unsent events to transmit.")
if state is not None:
commit_state(state, state_path)
return
try:
send_encrypted_logs_over_socket(config, candidate_logs)
resp = send_encrypted_logs_over_socket(config, candidate_logs)
if state is not None and resp and resp.get("status") == "success":
commit_state(state, state_path)
print(f"[+] State successfully committed to {state_path}")
except Exception as e:
print(f"[!] Failed to stream logs to server: {e}")
sys.exit(1)
+139 -56
View File
@@ -14,6 +14,7 @@
7. [Repository & Shippables Structure](#repository--shippables-structure)
8. [Getting Started & Installation](#getting-started--installation)
9. [Running Tests](#running-tests)
10. [Automated Releases via Gitea Actions](#automated-releases-via-gitea-actions)
---
@@ -22,7 +23,7 @@
### 1. Edge Thinness & Zero State
Site agents running on Windows and Linux act strictly as lightweight forwarders:
- **No Local Database**: Clients maintain zero state and no local SQLite or cache files.
- **Source-Level Noise Stripping**: Conversational, informational, and debugging log noise (`INFO`, `DEBUG`, audit entries) is dropped directly at the source.
- **Source-Level Filtering**: Agents upload candidate entries spanning from informational events up to errors (`INFO`, `WARNING`, `ERROR`, `CRITICAL`), while stripping verbose debugging noise (`DEBUG`, audit entries) and skipping any entries older than 24 hours.
- **End-to-End Encryption**: Logs are encrypted using the server's OpenPGP public key before leaving the edge node.
- **Secure TCP Sockets**: Ingestion occurs over low-overhead authenticated TCP sockets rather than bulky HTTP/HTTPS endpoints.
@@ -41,9 +42,9 @@ Instead of human engineers manually diving through noisy logs, **Hermes** ingest
```mermaid
graph TB
subgraph Edge Nodes [Zero-State Edge Forwarders]
W[Win_Client.py / Win_Client.exe<br/>Windows Event Log Application]
L[Linux_Client.py / Linux_Client.bin<br/>systemd journalctl -p warning]
subgraph Edge Nodes [State-Tracking Edge Forwarders]
W[Win_Client.py / Win_Client.exe / Win_Client.pyz<br/>Windows Event Log Application]
L[Linux_Client.py / Linux_Client.bin<br/>systemd journalctl -p info]
end
subgraph Security Layer [Security & Framing]
@@ -171,36 +172,32 @@ The server automatically infers site attribution from domain qualifiers (e.g. `n
```
LOGAR/
├── .gitea/
│ └── workflows/
│ ├── ci.yml # Continuous Integration automated test suite (runs on every push)
│ ├── release-linux.yml # Linux release workflow (compiles Server.bin, Linux_Client.bin, checksums)
│ └── release-windows.yml # Windows release workflow (compiles Server.exe, Win_Client.exe, checksums)
├── .gitignore # Ignore venv, caches, DBs, and private keys
├── requirements.txt # Unified dependencies
├── README.md # Comprehensive documentation
├── Server.py # Central TCP server and Hermes API
├── Win_Client.py # Windows edge forwarder
├── Linux_Client.py # Linux edge forwarder
├── server_config.sample.json # Central server sample configuration
├── package_dist.py # Multi-platform standalone binary packaging script
├── upload_release.py # Direct Gitea REST API release asset publisher
├── test_pipeline.py # End-to-end integration test
── out/ # Standalone shippable distributions
├── server/
│ ├── Server.py # Python source
── server_config.sample.json
│ ├── requirements.txt
│ ├── README.md
│ └── test/
│ └── test_server.py # Server unit tests
── tests/ # Unified unit test suites
├── test_server.py # Server unit tests
│ ├── test_win_client.py # Windows client unit tests
── test_linux_client.py # Linux client unit tests
└── out/ # Edge forwarder deployment packages
├── win_client/
│ ├── Win_Client.py # Python source
── client_config.sample.json
│ ├── requirements.txt
│ ├── README.md
│ └── test/
│ └── test_win_client.py # Windows client unit tests
│ ├── client_config.sample.json # Reference client configuration
── README.md # Windows service installation & configuration guide
└── linux_client/
├── build_bin.sh # PyInstaller native ELF compiler script
── Linux_Client.py # Python source
├── client_config.sample.json
├── requirements.txt
├── README.md
└── test/
└── test_linux_client.py# Linux client unit tests
├── client_config.sample.json # Reference client configuration
── README.md # Linux service installation & configuration guide
```
---
@@ -224,19 +221,48 @@ LOGAR/
### 2. Windows Client Deployment
1. Copy `Win_Client.py` (and `requirements.txt`) plus `client_config.json` to the target machine.
2. Run manually or schedule via Task Scheduler (every 3 hours):
#### Option A: Precompiled Standalone Executable (Recommended)
1. Download `Win_Client.exe` (or `Win_Client.pyz`) from the repository releases.
2. Place `client_config.json` (exported from the server) in the same directory.
3. Run manually or schedule via Task Scheduler (every 3 hours):
```powershell
python Win_Client.py --hours 6
.\Win_Client.exe --hours 24
```
#### Option B: Python Source Execution
1. Copy `Win_Client.py`, `requirements.txt`, and `client_config.json` to the target machine.
2. Install client dependencies:
```powershell
python -m pip install -r requirements.txt
```
3. Run manually or schedule via Task Scheduler:
```powershell
python Win_Client.py --hours 24
```
### 3. Linux Client Deployment
1. Copy `Linux_Client.py` (and `requirements.txt`) plus `client_config.json` to `/opt/logar/`.
2. (Optional) Run `build_bin.sh` to compile a standalone ELF binary if desired.
#### Option A: Precompiled Standalone Binary (Recommended)
1. Download `Linux_Client.bin` from the repository releases.
2. Place `Linux_Client.bin` and `client_config.json` into `/opt/logar/` and make it executable:
```bash
chmod +x /opt/logar/Linux_Client.bin
```
3. Run via cron or systemd timer:
```bash
0 */3 * * * python3 /opt/logar/Linux_Client.py --hours 6
0 */3 * * * /opt/logar/Linux_Client.bin --hours 24
```
#### Option B: Python Source Execution
1. Copy `Linux_Client.py`, `requirements.txt`, and `client_config.json` to `/opt/logar/`.
2. Install client dependencies:
```bash
python3 -m pip install -r requirements.txt
```
3. (Optional) Run `out/linux_client/build_bin.sh` to compile a standalone ELF binary locally if desired.
4. Run via cron or systemd timer:
```bash
0 */3 * * * python3 /opt/logar/Linux_Client.py --hours 24
```
---
@@ -244,46 +270,103 @@ LOGAR/
## Running Tests
### 1. Component-Specific Unit Tests
Each component in `out/` includes its own isolated test suite:
The test suite is located in `tests/` and exercises all components:
```bash
# Server tests (config generation, SQLite persistence, 4-run rule)
python out/server/test/test_server.py
# Run all unit tests
python -m unittest discover -s tests
# Windows client tests (config anonymity, machine ID, OpenPGP encryption)
python out/win_client/test/test_win_client.py
# Linux client tests (config anonymity, journalctl priority filter, OpenPGP)
python out/linux_client/test/test_linux_client.py
# Or run component tests individually:
python -m unittest tests/test_server.py
python -m unittest tests/test_win_client.py
python -m unittest tests/test_linux_client.py
```
### 2. End-to-End Pipeline Integration Test
Start the server in one shell and run the pipeline test:
```bash
python test_pipeline.py
```
This tests invalid token rejection, encrypted socket streaming, database persistence, status promotion upon the 4th run, and the Hermes API output.
The pipeline test exercises invalid token rejection, encrypted socket streaming, database persistence, status promotion upon the 4th run, and the Hermes API report output.
1. **Start the server** in Shell 1 (creates `server_config.json` on first run):
```bash
python Server.py
```
2. **Export client configuration** in Shell 2 (required for testing):
```bash
python Server.py --create-client-config --server-host 127.0.0.1 --server-port 9443 --client-out client_config.json
```
3. **Execute the integration test** in Shell 2:
```bash
python test_pipeline.py
```
## Continuous Integration via Gitea Actions
Continuous integration is automated via [`.gitea/workflows/ci.yml`](.gitea/workflows/ci.yml) and triggers automatically on **every push** and pull request:
1. **Syntax Compilation**: Validates all Python scripts (`Server.py`, `Win_Client.py`, `Linux_Client.py`, `package_dist.py`, `upload_release.py`, `test_pipeline.py`, and test suites).
2. **Component Unit Tests**: Discovers and runs all unit tests in `tests/` (`test_server.py`, `test_win_client.py`, `test_linux_client.py`).
3. **End-to-End Pipeline Verification**: Automatically spins up the LOGAR server hub, generates test configs, runs `test_pipeline.py` (testing socket authentication, 4-run rule persistence, Hermes API report, and client integrations), and shuts down the test instance.
---
## Automated Releases via Gitea Actions
Releases are automated via [`.gitea/workflows/release.yml`](.gitea/workflows/release.yml) using your Gitea action runner:
Release builds are automated via two dedicated Gitea Actions workflows running concurrently on native platform runners:
- [`.gitea/workflows/release-linux.yml`](.gitea/workflows/release-linux.yml) (`ubuntu-latest`)
- [`.gitea/workflows/release-windows.yml`](.gitea/workflows/release-windows.yml) (`windows-latest`)
### Publishing a Release
Whenever you want to release a new version with compiled standalone binaries:
```bash
git tag v1.0.0
git push origin v1.0.0
git tag v1.0.1
git push origin v1.0.1
```
*(You can also trigger builds manually via the Gitea UI using the **Run workflow** button (`workflow_dispatch`) on either workflow).*
### What Gitea Actions Does Automatically:
1. Gitea runner executes the workflow on tag push.
2. Runs `package_dist.py` to compile native standalone binaries:
- `Linux_Client.bin` (standalone binary)
- `Server.bin` (standalone server binary)
- `Win_Client.pyz` (standalone executable zipapp)
- `SHA256SUMS.txt` (checksums)
3. Publishes the Gitea release using `gitea-release-action` and attaches the compiled binary assets.
### Automated Multi-Platform Compilation:
1. **Linux Runner** (`release-linux.yml`):
- Compiles native Linux ELF executables: `Linux_Client.bin` and `Server.bin`.
- Generates dedicated SHA-256 checksum files:
- `linux_client_sha256sum` (verification for `Linux_Client.bin`)
- `linux_agent_sha256sum` (alias for client/agent integrations)
- `linux_server_sha256sum` (verification for `Server.bin`)
- `SHA256SUMS_linux.txt` (summary manifest)
- Attaches all Linux assets to the Gitea release.
*(Note: You can also use `upload_release.py` from your Windows machine to upload Windows `.exe` binaries directly if desired).*
2. **Windows Runner** (`release-windows.yml`):
- Compiles native Windows PE executables: `Win_Client.exe` and `Server.exe`.
- Generates dedicated SHA-256 checksum files:
- `win_client_sha256sum` (verification for `Win_Client.exe`)
- `win_agent_sha256sum` (alias for client/agent integrations)
- `win_server_sha256sum` (verification for `Server.exe`)
- `SHA256SUMS_windows.txt` (summary manifest)
- Attaches all Windows assets to the Gitea release.
3. **Concurrent Publishing & Conflict Handling**:
`upload_release.py` includes automatic retry and conflict resolution so concurrent Windows and Linux runners attach their respective assets to the release without collision.
### Verifying Checksums
- On Linux:
```bash
sha256sum -c linux_client_sha256sum
# or
sha256sum -c linux_server_sha256sum
```
- On Windows (PowerShell):
```powershell
Get-FileHash .\Win_Client.exe -Algorithm SHA256
Get-Content .\win_client_sha256sum
```
### Local Packaging & Manual Upload
You can also compile and package binaries locally anytime:
```bash
# Windows
py -3.12 package_dist.py --target windows
# Linux
python3 package_dist.py --target linux
```
To upload local builds directly to Gitea:
```powershell
python upload_release.py --tag v1.0.1 --token <YOUR_GITEA_TOKEN>
```
*(Environment variables `GITEA_TOKEN`, `GITEA_SERVER_URL`, `GITEA_REPOSITORY`, and `GITEA_REF_NAME` are also supported automatically).*
+10
View File
@@ -0,0 +1,10 @@
# LOGAR Release v1.0.1
### Changes in this Release:
- **Dual Platform Gitea Release Automation**: Added dedicated Windows (`release-windows.yml`) and Linux (`release-linux.yml`) Gitea Actions to compile native platform binaries (`Win_Client.exe` and `Server.exe` on Windows; `Linux_Client.bin` and `Server.bin` on Linux).
- **Dedicated SHA-256 Checksums**: Release assets now include dedicated checksum files matching `[win/linux]_[client/agent]_sha256sum` (`win_client_sha256sum`, `win_agent_sha256sum`, `win_server_sha256sum`, `linux_client_sha256sum`, `linux_agent_sha256sum`, `linux_server_sha256sum`).
- **Removed Client Filter Logic**: Removed restrictive source-level noise filtering on edge forwarders. Clients now collect and stream all candidate events from `INFO` up to `ERROR` over the lookback window instead of discarding them at the source.
- **State Tracking & Deduplication**: Added persistent client state tracking (`client_state.json`) with journalctl cursors and Windows Event Log record numbers to guarantee that previously transmitted events are never resent.
- **24-Hour Lookback Window**: Forwarders now scan and upload events from the last 24 hours (default `--hours 24`), skipping older entries.
- **Lightweight Distribution Structure**: Cleaned `out/` to strictly contain deployment documentation and sample configurations.
- **Automated Gitea CI/CD**: Integrated push testing workflow (`ci.yml`) and multi-platform release asset packaging.
+2 -2
View File
@@ -179,8 +179,8 @@ def process_ingested_logs(payload: Dict[str, Any], db_path: str, window_hours: i
for log in logs:
severity = str(log.get("severity", "WARNING")).upper()
# Edge forwarder filter safeguard (strip informational noise)
if severity in ["INFO", "DEBUG"]:
# Edge forwarder filter safeguard: retain INFO to ERROR / CRITICAL; strip verbose debug noise
if severity in ["DEBUG", "TRACE"]:
continue
signature = log.get("signature", "unknown")
+114 -13
View File
@@ -6,6 +6,7 @@ import struct
import argparse
import warnings
from datetime import datetime, timezone, timedelta
from typing import Optional, Dict, Any, List
# Suppress cryptography / pgpy deprecation notices
warnings.filterwarnings("ignore")
@@ -18,6 +19,45 @@ except ImportError:
win32evtlog = None
CONFIG_FILE_NAME = "client_config.json"
STATE_FILE_NAME = "client_state.json"
def get_state_path(config_path: str, custom_state_path: Optional[str] = None) -> str:
if custom_state_path:
return custom_state_path
config_dir = os.path.dirname(os.path.abspath(config_path))
return os.path.join(config_dir, STATE_FILE_NAME)
def load_state(state_path: str) -> dict:
if os.path.exists(state_path):
try:
with open(state_path, "r", encoding="utf-8") as f:
return json.load(f)
except Exception as e:
print(f"[!] Warning: Failed to read state file '{state_path}': {e}")
return {}
return {}
def save_state(state_path: str, state: dict):
try:
temp_path = f"{state_path}.tmp"
with open(temp_path, "w", encoding="utf-8") as f:
json.dump(state, f, indent=2)
os.replace(temp_path, state_path)
except Exception as e:
print(f"[!] Warning: Could not save client state to '{state_path}': {e}")
def commit_state(state: dict, state_path: str):
if "new_last_record_number" in state:
val = state.pop("new_last_record_number")
if val:
state["last_record_number"] = val
if "new_sent_record_ids" in state:
state["sent_record_ids"] = state.pop("new_sent_record_ids")
save_state(state_path, state)
def load_config(config_path: str = CONFIG_FILE_NAME):
@@ -55,10 +95,12 @@ def get_machine_identifier() -> str:
return hostname
def get_recent_windows_logs(hours: int = 6):
def get_recent_windows_logs(hours: int = 24, state: Optional[dict] = None) -> list:
"""
Scans the Windows Application Event Log backwards for events within the window.
Edge Thinness & Noise Stripping: INFO and DEBUG events are dropped at the source.
Edge Filtering: Retains INFO, WARNING, and ERROR. Drops Audit and Debug noise.
State Tracking: Skips events older than lookback window (default 24h) and events
that have already been sent in previous runs.
"""
if win32evtlog is None:
print("[!] pywin32 is not installed or not running on Windows. Returning mock/empty candidate list.")
@@ -78,23 +120,51 @@ def get_recent_windows_logs(hours: int = 6):
cutoff_time = datetime.now() - timedelta(hours=hours)
machine_id = get_machine_identifier()
last_record_number = 0
sent_record_ids = set()
if state:
last_record_number = int(state.get("last_record_number", 0))
sent_record_ids = set(state.get("sent_record_ids", []))
# Windows Event Log EventTypes:
# 1: EVENTLOG_ERROR_TYPE -> ERROR
# 2: EVENTLOG_WARNING_TYPE -> WARNING
# 4: EVENTLOG_INFORMATION_TYPE -> INFO
# Excludes: 8 (Audit Success), 16 (Audit Failure), and other verbose noise
sev_map = {
1: "CRITICAL",
2: "ERROR",
3: "WARNING"
1: "ERROR",
2: "WARNING",
4: "INFO"
}
newest_record_number = 0
collected_record_ids = []
while True:
events = win32evtlog.ReadEventLog(hand, flags, 0)
if not events:
break
for event in events:
rec_num = int(event.RecordNumber)
if newest_record_number == 0:
newest_record_number = rec_num
# 1. Skip entries older than lookback window (default: 24h)
if event.TimeGenerated < cutoff_time:
break
# Drop conversational or informational noise (INFO=4, etc.) at source
# Only retain Critical (1), Error (2), and Warning (3)
# 2. Skip already sent events if we've reached records <= last_record_number
# (unless the log was cleared and numbers wrapped, i.e. newest_record_number < last_record_number)
if last_record_number > 0 and newest_record_number >= last_record_number:
if rec_num <= last_record_number:
break
rec_id = f"{rec_num}:{event.TimeGenerated.isoformat()}"
if rec_id in sent_record_ids:
continue
# Filter: upload everything from INFO to ERROR only
if event.EventType in sev_map:
msg = " ".join(event.StringInserts) if event.StringInserts else "Event Log Entry"
logs.append({
@@ -104,11 +174,21 @@ def get_recent_windows_logs(hours: int = 6):
"severity": sev_map[event.EventType],
"message": msg[:2048] # Limit message length
})
collected_record_ids.append(rec_id)
if events[-1].TimeGenerated < cutoff_time:
break
if last_record_number > 0 and newest_record_number >= last_record_number and events[-1].RecordNumber <= last_record_number:
break
win32evtlog.CloseEventLog(hand)
if state is not None:
target_rec = max(newest_record_number, last_record_number)
state["new_last_record_number"] = target_rec
state["new_sent_record_ids"] = (list(sent_record_ids) + collected_record_ids)[-1000:]
state["last_run_timestamp"] = datetime.now(timezone.utc).isoformat()
return logs
@@ -183,9 +263,11 @@ def send_encrypted_logs_over_socket(config: dict, logs: list):
def main():
parser = argparse.ArgumentParser(description="LOGAR Windows Edge Log Forwarder (Zero State)")
parser = argparse.ArgumentParser(description="LOGAR Windows Edge Log Forwarder with State Tracking")
parser.add_argument("--config", default=CONFIG_FILE_NAME, help="Path to client_config.json")
parser.add_argument("--hours", type=int, default=6, help="Lookback window in hours for event logs")
parser.add_argument("--hours", type=int, default=24, help="Lookback window in hours for event logs (default: 24)")
parser.add_argument("--state-file", default=None, help="Path to state tracking file (default: client_state.json next to config)")
parser.add_argument("--no-state", action="store_true", help="Disable state tracking and send all events matching lookback window")
args = parser.parse_args()
try:
@@ -194,12 +276,31 @@ def main():
print(f"[!] Configuration error: {e}")
sys.exit(1)
print(f"[*] Scanning Windows Application event log for candidate anomalies (last {args.hours} hours)...")
candidate_logs = get_recent_windows_logs(hours=args.hours)
print(f"[*] Found {len(candidate_logs)} candidate anomalies (noise stripped at source).")
state_path = get_state_path(args.config, args.state_file)
state = None if args.no_state else load_state(state_path)
machine_id = get_machine_identifier()
print(f"[*] Edge Forwarder Node: {machine_id}")
if state and "last_record_number" in state:
print(f"[*] State tracking active: resuming from record #{state['last_record_number']} (state file: {state_path})")
elif not args.no_state:
print(f"[*] State tracking initialized (state file: {state_path})")
print(f"[*] Scanning Windows Application event log for unsent entries (last {args.hours} hours)...")
candidate_logs = get_recent_windows_logs(hours=args.hours, state=state)
print(f"[*] Found {len(candidate_logs)} unsent candidate entries (INFO to ERROR, entries > {args.hours}h and already-sent skipped).")
if not candidate_logs:
print("[*] No new unsent events to transmit.")
if state is not None:
commit_state(state, state_path)
return
try:
send_encrypted_logs_over_socket(config, candidate_logs)
resp = send_encrypted_logs_over_socket(config, candidate_logs)
if state is not None and resp and resp.get("status") == "success":
commit_state(state, state_path)
print(f"[+] State successfully committed to {state_path}")
except Exception as e:
print(f"[!] Failed to stream logs to server: {e}")
sys.exit(1)
-194
View File
@@ -1,194 +0,0 @@
import os
import sys
import json
import socket
import struct
import argparse
import subprocess
import warnings
from datetime import datetime, timezone
# Suppress cryptography / pgpy deprecation notices
warnings.filterwarnings("ignore")
import pgpy
CONFIG_FILE_NAME = "client_config.json"
def load_config(config_path: str = CONFIG_FILE_NAME):
if not os.path.exists(config_path):
raise FileNotFoundError(
f"Client configuration file not found at: {config_path}\n"
f"Generate one from the server using: python Server.py --create-client-config --client-out {config_path}"
)
with open(config_path, "r", encoding="utf-8") as f:
return json.load(f)
def get_machine_identifier() -> str:
"""
Returns the hostname of the machine sending the logs,
and appends the network/DNS domain if available.
"""
fqdn = socket.getfqdn()
if fqdn and "." in fqdn and not fqdn.startswith("localhost"):
return fqdn
hostname = socket.gethostname()
try:
if os.path.exists("/etc/resolv.conf"):
with open("/etc/resolv.conf", "r", encoding="utf-8") as f:
for line in f:
parts = line.strip().split()
if parts and parts[0] in ["domain", "search"] and len(parts) > 1:
domain = parts[1]
if domain and not domain.startswith("."):
return f"{hostname}.{domain}"
except Exception:
pass
try:
host_ip = socket.gethostbyname(hostname)
canonical_name = socket.gethostbyaddr(host_ip)[0]
if canonical_name and "." in canonical_name and not canonical_name.startswith("localhost"):
return canonical_name
except Exception:
pass
return hostname
def get_recent_linux_logs(hours: int = 6):
"""
Collects warnings and errors from systemd journalctl over the lookback window.
Edge Thinness: Drops INFO and DEBUG entries at the source.
"""
cmd = ["journalctl", "--since", f"{hours} hours ago", "-p", "warning", "--output=json"]
try:
result = subprocess.run(cmd, capture_output=True, text=True, check=False)
except FileNotFoundError:
print("[!] journalctl command not found. Ensure this script runs on a systemd-enabled Linux system.")
return []
except Exception as e:
print(f"[!] Error running journalctl: {e}")
return []
logs = []
machine_id = get_machine_identifier()
for line in result.stdout.splitlines():
line_str = line.strip()
if not line_str:
continue
try:
entry = json.loads(line_str)
priority = str(entry.get("PRIORITY", "4"))
if int(priority) > 4:
continue
sev = "WARNING" if priority == "4" else "ERROR"
logs.append({
"server": machine_id,
"os_type": "linux",
"signature": entry.get("SYSLOG_IDENTIFIER", "unknown"),
"severity": sev,
"message": entry.get("MESSAGE", "")[:2048]
})
except (json.JSONDecodeError, ValueError):
continue
return logs
def send_encrypted_logs_over_socket(config: dict, logs: list):
"""
Encrypts the payload using the server's OpenPGP public key and streams
over an authenticated TCP socket. Zero local state is maintained on the client.
"""
server_host = config["server_host"]
server_port = int(config["server_port"])
auth_token = config["auth_token"]
pub_key_armored = config["server_public_key"]
expected_fp = config.get("server_fingerprint", "").replace(" ", "").upper()
pub_key, _ = pgpy.PGPKey.from_blob(pub_key_armored)
actual_fp = str(pub_key.fingerprint).replace(" ", "").upper()
if expected_fp and actual_fp != expected_fp:
raise ValueError(
f"Server fingerprint mismatch! Expected {expected_fp}, but key has {actual_fp}."
)
machine_id = get_machine_identifier()
payload = {
"server": machine_id,
"timestamp": datetime.now(timezone.utc).isoformat(),
"logs": logs
}
payload_json = json.dumps(payload)
pgp_msg = pgpy.PGPMessage.new(payload_json)
encrypted_msg = pub_key.encrypt(pgp_msg)
encrypted_armored = str(encrypted_msg)
envelope = {
"auth_token": auth_token,
"timestamp": datetime.now(timezone.utc).isoformat(),
"encrypted_payload": encrypted_armored
}
envelope_bytes = json.dumps(envelope).encode("utf-8")
print(f"[*] Connecting to LOGAR server at {server_host}:{server_port} over secure TCP socket...")
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
sock.settimeout(15.0)
sock.connect((server_host, server_port))
frame = struct.pack(">I", len(envelope_bytes)) + envelope_bytes
sock.sendall(frame)
resp_len_bytes = sock.recv(4)
if not resp_len_bytes:
raise ConnectionError("Server closed connection without response.")
resp_len = struct.unpack(">I", resp_len_bytes)[0]
resp_bytes = bytearray()
while len(resp_bytes) < resp_len:
chunk = sock.recv(min(4096, resp_len - len(resp_bytes)))
if not chunk:
break
resp_bytes.extend(chunk)
response = json.loads(resp_bytes.decode("utf-8"))
print(f"[+] Server response: {response}")
return response
def main():
parser = argparse.ArgumentParser(description="LOGAR Linux Edge Log Forwarder (Zero State)")
parser.add_argument("--config", default=CONFIG_FILE_NAME, help="Path to client_config.json")
parser.add_argument("--hours", type=int, default=6, help="Lookback window in hours for journalctl logs")
args = parser.parse_args()
try:
config = load_config(args.config)
except Exception as e:
print(f"[!] Configuration error: {e}")
sys.exit(1)
machine_id = get_machine_identifier()
print(f"[*] Edge Forwarder Node: {machine_id}")
print(f"[*] Scanning Linux journalctl for candidate anomalies (last {args.hours} hours)...")
candidate_logs = get_recent_linux_logs(hours=args.hours)
print(f"[*] Found {len(candidate_logs)} candidate anomalies (noise stripped at source).")
try:
send_encrypted_logs_over_socket(config, candidate_logs)
except Exception as e:
print(f"[!] Failed to stream logs to server: {e}")
sys.exit(1)
if __name__ == "__main__":
main()
+112 -25
View File
@@ -1,50 +1,111 @@
# LOGAR Linux Edge Forwarder
Lightweight edge log forwarder for Linux servers running systemd.
Standalone compiled binary distribution for Linux edge servers running systemd.
## Features
- **Zero Local State**: No local SQLite database or state tracking on the edge server.
- **Edge Noise Stripping**: Strips conversational/informational noise (`INFO`, `DEBUG`) directly at the source via `journalctl -p warning`.
- **End-to-End OpenPGP Encryption**: Encrypts logs using the server's public key; decrypted exclusively on the cloud hub.
- **Authenticated TCP Socket**: Direct, low-overhead TCP streaming with token authentication.
- **No GPG Binary Required**: Pure-Python implementation (`pgpy` + `cryptography`).
---
## Overview
`Linux_Client.bin` is a self-contained, pre-compiled executable that queries `systemd-journald` via `journalctl`, filters logs directly at the source, encrypts the payload using OpenPGP, and streams candidate events over an authenticated TCP socket to the central LOGAR hub.
### Key Capabilities
- **Pre-compiled & Dependency-Free**: Ships as a standalone executable binary (`Linux_Client.bin`). No Python environment, pip packages, or GnuPG binaries are required on the host.
- **Source-Level Filtering**: Retains events spanning `INFO`, `WARNING`, and `ERROR` (`journalctl -p info`). Drops debug noise (priority 7) and skips events older than 24 hours.
- **State Tracking & Deduplication**: Maintains persistent client state in `client_state.json` (tracking systemd journalctl cursors and microsecond timestamps) so every log record is forwarded exactly once without duplicates.
- **Fail-Safe State Commit**: State is committed only when the server returns a verified `success` response. In the event of a network outage, state remains unchanged and unsent events are retried automatically on the next run.
- **End-to-End Encryption**: Encrypts payloads using the server's OpenPGP public key before transmission.
---
## 1. Generating & Deploying the Configuration File
### Step 1: Generate `client_config.json` on the Server
Run the following command on your central LOGAR server to export a client bundle tailored for your environment:
## Installation
```bash
python3 -m pip install -r requirements.txt
python Server.py --create-client-config --server-host <SERVER_IP_OR_DNS> --server-port 9443 --client-out client_config.json
```
## Configuration
Place `client_config.json` generated by the server (`Server.py --create-client-config`) in the same directory as `Linux_Client.py`.
- Replace `<SERVER_IP_OR_DNS>` with the reachable IP address or FQDN of your central LOGAR server hub.
- Default TCP port is `9443`.
## Running the Forwarder
```bash
python3 Linux_Client.py --hours 6
### Step 2: Configuration Structure
The generated `client_config.json` contains:
```json
{
"server_host": "192.168.1.100",
"server_port": 9443,
"server_fingerprint": "375388960531264EA0648EC0D2C4E4ABC6F22AC2",
"server_public_key": "-----BEGIN PGP PUBLIC KEY BLOCK-----\n...",
"auth_token": "a1b2c3d4e5f6..."
}
```
## Cron / Systemd Timer Deployment
### Option A: Cron Job (Every 3 hours)
> [!NOTE]
> A reference example is provided in `client_config.sample.json`. The configuration file contains **no host-specific names or site names** to ensure client anonymity and easy redistribution.
### Step 3: Copy to Edge Node
Place `Linux_Client.bin` and `client_config.json` into the target directory (recommended: `/opt/logar/`):
```bash
0 */3 * * * cd /opt/logar && /usr/bin/python3 Linux_Client.py --hours 6 >> /var/log/logar_client.log 2>&1
sudo mkdir -p /opt/logar
sudo cp Linux_Client.bin client_config.json /opt/logar/
sudo chmod +x /opt/logar/Linux_Client.bin
```
### Option B: Systemd Service & Timer
1. Create `/etc/systemd/system/logar-forwarder.service`:
---
## 2. Running Manually
Test the forwarder interactively:
```bash
cd /opt/logar
./Linux_Client.bin --hours 24
```
### Command-Line Arguments
| Argument | Default | Description |
| :--- | :--- | :--- |
| `--config` | `client_config.json` | Path to client configuration file |
| `--hours` | `24` | Lookback window in hours for journal logs |
| `--state-file` | `client_state.json` | Path to persistent state file |
| `--no-state` | `False` | Disable state tracking and send all events matching lookback window |
---
## 3. Installing as a Systemd Service & Timer (Recommended)
Running `Linux_Client.bin` via a systemd timer ensures reliable periodic execution, automatic restart, and native log integration with `journalctl`.
### Step 1: Create the Systemd Service Unit
Create `/etc/systemd/system/logar-forwarder.service`:
```ini
[Unit]
Description=LOGAR Edge Forwarder
After=network.target
Description=LOGAR Edge Log Forwarder
After=network-online.target
Wants=network-online.target
[Service]
Type=oneshot
WorkingDirectory=/opt/logar
ExecStart=/usr/bin/python3 /opt/logar/Linux_Client.py --hours 6
ExecStart=/opt/logar/Linux_Client.bin --hours 24
User=root
StandardOutput=journal
StandardError=journal
[Install]
WantedBy=multi-user.target
```
2. Create `/etc/systemd/system/logar-forwarder.timer`:
### Step 2: Create the Systemd Timer Unit
Create `/etc/systemd/system/logar-forwarder.timer` to execute the forwarder every 3 hours (with a 5-minute initial delay upon boot):
```ini
[Unit]
Description=Run LOGAR Edge Forwarder every 3 hours
Description=Run LOGAR Edge Forwarder periodically
Requires=logar-forwarder.service
[Timer]
OnBootSec=5min
@@ -55,8 +116,34 @@ Persistent=true
WantedBy=timers.target
```
3. Enable and start:
### Step 3: Enable and Start the Timer
```bash
sudo systemctl daemon-reload
sudo systemctl enable --now logar-forwarder.timer
```
### Step 4: Verify Timer & Service Status
```bash
# Check timer schedule
sudo systemctl list-timers --all | grep logar
# Trigger an immediate manual execution
sudo systemctl start logar-forwarder.service
# View execution logs
sudo journalctl -u logar-forwarder.service -n 50
```
---
## 4. Alternative: Cron Job Deployment
If systemd timers are not preferred, configure a periodic cron job running every 3 hours:
```bash
# Open root crontab
sudo crontab -e
# Add the following entry:
0 */3 * * * cd /opt/logar && ./Linux_Client.bin --hours 24 >> /var/log/logar_forwarder.log 2>&1
```
-11
View File
@@ -1,11 +0,0 @@
#!/usr/bin/env bash
# Build script to compile Linux_Client into a standalone native ELF binary on Linux
set -e
echo "[*] Installing build requirements..."
pip3 install pyinstaller pgpy cryptography standard-imghdr
echo "[*] Compiling Linux_Client native binary..."
pyinstaller --onefile --clean --name Linux_Client.bin Linux_Client.py
echo "[+] Compilation successful: dist/Linux_Client.bin"
+1 -2
View File
@@ -3,6 +3,5 @@
"server_port": 9443,
"server_fingerprint": "PASTE_SERVER_FINGERPRINT_HERE",
"server_public_key": "-----BEGIN PGP PUBLIC KEY BLOCK-----\n...\n-----END PGP PUBLIC KEY BLOCK-----\n",
"auth_token": "PASTE_AUTH_TOKEN_HERE",
"site_name": "Frankfurt-DC"
"auth_token": "PASTE_AUTH_TOKEN_HERE"
}
-3
View File
@@ -1,3 +0,0 @@
pgpy>=0.6.0
standard-imghdr>=3.13.0; python_version >= "3.13"
cryptography>=42.0.0
-107
View File
@@ -1,107 +0,0 @@
import os
import sys
import json
import struct
import unittest
import warnings
warnings.filterwarnings("ignore")
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
import Linux_Client
import pgpy
from pgpy.constants import PubKeyAlgorithm, KeyFlags, HashAlgorithm, SymmetricKeyAlgorithm, CompressionAlgorithm
class TestLinuxClientComponent(unittest.TestCase):
def setUp(self):
self.dummy_config = "test_linux_client_config.json"
# Generate dummy PGP key for testing
key = pgpy.PGPKey.new(PubKeyAlgorithm.RSAEncryptOrSign, 2048)
uid = pgpy.PGPUID.new("TestHub")
key.add_uid(
uid,
usage={KeyFlags.EncryptCommunications, KeyFlags.EncryptStorage},
hashes=[HashAlgorithm.SHA256],
ciphers=[SymmetricKeyAlgorithm.AES256],
compression=[CompressionAlgorithm.Uncompressed]
)
self.server_priv = key
self.server_pub = key.pubkey
self.fingerprint = str(key.pubkey.fingerprint)
with open(self.dummy_config, "w", encoding="utf-8") as f:
json.dump({
"server_host": "127.0.0.1",
"server_port": 9443,
"server_fingerprint": self.fingerprint,
"server_public_key": str(self.server_pub),
"auth_token": "secret-test-token"
}, f)
def tearDown(self):
if os.path.exists(self.dummy_config):
try:
os.remove(self.dummy_config)
except Exception:
pass
def test_client_config_anonymity(self):
config = Linux_Client.load_config(self.dummy_config)
self.assertNotIn("server_name", config)
self.assertNotIn("name", config)
self.assertNotIn("site_name", config)
self.assertEqual(config["server_fingerprint"], self.fingerprint)
def test_get_machine_identifier(self):
machine_id = Linux_Client.get_machine_identifier()
self.assertIsInstance(machine_id, str)
self.assertGreater(len(machine_id), 0)
self.assertNotEqual(machine_id, "localhost")
def test_journalctl_parsing_and_priority_filter(self):
sample_journal_lines = [
json.dumps({"PRIORITY": "3", "SYSLOG_IDENTIFIER": "sshd", "MESSAGE": "Failed password for root"}),
json.dumps({"PRIORITY": "4", "SYSLOG_IDENTIFIER": "systemd", "MESSAGE": "Unit entered failed state"}),
json.dumps({"PRIORITY": "6", "SYSLOG_IDENTIFIER": "cron", "MESSAGE": "Informational session opened"}),
]
logs = []
machine_id = Linux_Client.get_machine_identifier()
for line_str in sample_journal_lines:
entry = json.loads(line_str)
priority = str(entry.get("PRIORITY", "4"))
if int(priority) > 4:
continue
sev = "WARNING" if priority == "4" else "ERROR"
logs.append({
"server": machine_id,
"os_type": "linux",
"signature": entry.get("SYSLOG_IDENTIFIER", "unknown"),
"severity": sev,
"message": entry.get("MESSAGE", "")
})
# Priority 6 must be stripped (INFO noise)
self.assertEqual(len(logs), 2)
self.assertEqual(logs[0]["severity"], "ERROR")
self.assertEqual(logs[1]["severity"], "WARNING")
def test_encryption_and_decryption(self):
config = Linux_Client.load_config(self.dummy_config)
pub_key, _ = pgpy.PGPKey.from_blob(config["server_public_key"])
payload = {
"server": Linux_Client.get_machine_identifier(),
"logs": [{"signature": "kernel", "severity": "ERROR", "message": "Kernel panic - not syncing"}]
}
msg = pgpy.PGPMessage.new(json.dumps(payload))
enc = pub_key.encrypt(msg)
self.assertTrue(str(enc).startswith("-----BEGIN PGP MESSAGE-----"))
dec = self.server_priv.decrypt(enc)
restored = json.loads(dec.message)
self.assertEqual(restored["logs"][0]["signature"], "kernel")
if __name__ == "__main__":
unittest.main()
-36
View File
@@ -1,36 +0,0 @@
# LOGAR Server Hub
Central Python/TCP ingestion server for the LOGAR Log Analysis System.
## Features
- **Zero External GPG Requirement**: Uses pure-Python OpenPGP (`pgpy` + `cryptography`), no native GnuPG binary needed.
- **First-Run Key & Config Auto-generation**: Generates OpenPGP keypairs, auth tokens, and `server_config.json` automatically on first launch.
- **Client Config Exporter**: Generates `client_config.json` bundles containing the server's encryption-only fingerprint and address.
- **Cloud-Side Temporal Persistence**: SQLite database tracking candidate anomalies over 12-hour evaluation windows.
- **4-Run Persistence Rule**: Filters out transient infrastructure blips, promoting issues to `VERIFIED` anomalies only after persisting across $\ge 4$ runs.
- **Agentic Hermes Endpoint**: REST API (`GET /api/hermes/report`) providing verified system artifacts for Hermes agent alerts.
## Installation
```bash
pip install -r requirements.txt
```
## Running the Server
```bash
# Starts both the TCP socket listener (port 9443) and the Hermes API (port 8443)
python Server.py
```
## Generating Client Configurations
To deploy edge forwarders, generate a client config file:
```bash
python Server.py --create-client-config --server-host <SERVER_IP_OR_DNS> --server-port 9443 --site-name "Frankfurt-DC" --client-out client_config.json
```
Copy the generated `client_config.json` into the deployment directory of `Win_Client.py` or `Linux_Client.py`.
## Hermes Agent Integration
Hermes queries the verified anomalies via:
```
GET http://<SERVER_IP>:8443/api/hermes/report
```
Only issues meeting the 4-run persistence rule within the active 12-hour evaluation window are returned.
-437
View File
@@ -1,437 +0,0 @@
# Copy of Server.py without site_name in client_config
import os
import sys
import json
import uuid
import struct
import socket
import sqlite3
import argparse
import asyncio
import secrets
import warnings
from datetime import datetime, timezone, timedelta
from typing import Dict, Any, List, Optional
# Suppress cryptography / pgpy deprecation notices for a clean terminal output
warnings.filterwarnings("ignore")
import pgpy
from pgpy.constants import (
PubKeyAlgorithm,
KeyFlags,
HashAlgorithm,
SymmetricKeyAlgorithm,
CompressionAlgorithm
)
from fastapi import FastAPI, HTTPException
import uvicorn
CONFIG_FILE_NAME = "server_config.json"
DEFAULT_DB_FILE = "logar_state.db"
EVALUATION_WINDOW_HOURS = 12
RUN_THRESHOLD = 4
app = FastAPI(title="LOGAR Cloud Ingestion & Hermes Hub", version="2.0.0")
SERVER_STATE: Dict[str, Any] = {}
def generate_server_keypair(server_name: str):
"""Generates an OpenPGP RSA 2048 key with encryption capability."""
key = pgpy.PGPKey.new(PubKeyAlgorithm.RSAEncryptOrSign, 2048)
uid = pgpy.PGPUID.new(server_name)
key.add_uid(
uid,
usage={KeyFlags.EncryptCommunications, KeyFlags.EncryptStorage},
hashes=[HashAlgorithm.SHA256],
ciphers=[SymmetricKeyAlgorithm.AES256],
compression=[CompressionAlgorithm.Uncompressed]
)
private_key_armored = str(key)
public_key_armored = str(key.pubkey)
fingerprint = str(key.pubkey.fingerprint)
return private_key_armored, public_key_armored, fingerprint
def load_or_init_config(config_path: str = CONFIG_FILE_NAME) -> Dict[str, Any]:
"""Loads existing server_config.json or creates a new one on first run."""
if os.path.exists(config_path):
print(f"[*] Loading server configuration from: {os.path.abspath(config_path)}")
with open(config_path, "r", encoding="utf-8") as f:
config = json.load(f)
return config
print(f"[!] Config '{config_path}' not found. Initializing first-run configuration...")
server_name = "LOGAR-Cloud-Hub"
private_key, public_key, fingerprint = generate_server_keypair(server_name)
auth_token = secrets.token_hex(24)
config = {
"server_name": server_name,
"tcp_host": "0.0.0.0",
"tcp_port": 9443,
"hermes_host": "0.0.0.0",
"hermes_port": 8443,
"auth_token": auth_token,
"db_path": DEFAULT_DB_FILE,
"evaluation_window_hours": EVALUATION_WINDOW_HOURS,
"min_persistence_runs": RUN_THRESHOLD,
"server_fingerprint": fingerprint,
"public_key": public_key,
"private_key": private_key
}
with open(config_path, "w", encoding="utf-8") as f:
json.dump(config, f, indent=2)
print(f"[+] Successfully generated new server config and OpenPGP keypair.")
print(f"[+] Server Encryption Fingerprint: {fingerprint}")
print(f"[+] Saved to: {os.path.abspath(config_path)}")
return config
def create_client_config(
server_host: str,
server_port: int,
output_path: str,
config_path: str = CONFIG_FILE_NAME
) -> Dict[str, Any]:
"""Creates a client configuration file containing the server address, auth token, and encryption-only key/fingerprint."""
server_conf = load_or_init_config(config_path)
client_conf = {
"server_host": server_host,
"server_port": server_port,
"server_fingerprint": server_conf["server_fingerprint"],
"server_public_key": server_conf["public_key"],
"auth_token": server_conf["auth_token"]
}
out_dir = os.path.dirname(os.path.abspath(output_path))
if out_dir and not os.path.exists(out_dir):
os.makedirs(out_dir, exist_ok=True)
with open(output_path, "w", encoding="utf-8") as f:
json.dump(client_conf, f, indent=2)
print(f"[+] Client configuration successfully written to: {os.path.abspath(output_path)}")
print(f" - Server Target: {server_host}:{server_port}")
print(f" - Encryption Fingerprint: {server_conf['server_fingerprint']}")
return client_conf
def init_db(db_path: str):
"""Initializes the SQLite schema for multi-run temporal tracking."""
conn = sqlite3.connect(db_path)
conn.execute("""
CREATE TABLE IF NOT EXISTS active_issues (
fingerprint TEXT PRIMARY KEY,
site_name TEXT,
server TEXT,
signature TEXT,
severity TEXT,
message TEXT,
os_type TEXT,
first_seen TEXT,
last_seen TEXT,
run_count INTEGER,
status TEXT,
last_run_id TEXT
)
""")
conn.execute("""
CREATE TABLE IF NOT EXISTS ingest_runs (
run_id TEXT PRIMARY KEY,
site_name TEXT,
server TEXT,
timestamp TEXT,
log_count INTEGER
)
""")
conn.commit()
conn.close()
def process_ingested_logs(payload: Dict[str, Any], db_path: str, window_hours: int, min_runs: int) -> Dict[str, Any]:
"""
Evaluates candidate issues against the 12-hour evaluation window and 4-run rule.
Zero-state clients send raw candidate entries; this engine handles temporal state.
"""
client_server = payload.get("server", "unknown-host")
site_name = payload.get("site_name") or (client_server.split(".", 1)[1] if "." in client_server else "default")
logs = payload.get("logs", [])
run_id = str(uuid.uuid4())
now = datetime.now(timezone.utc)
now_iso = now.isoformat()
conn = sqlite3.connect(db_path)
cursor = conn.cursor()
cursor.execute(
"INSERT INTO ingest_runs (run_id, site_name, server, timestamp, log_count) VALUES (?, ?, ?, ?, ?)",
(run_id, site_name, client_server, now_iso, len(logs))
)
processed_count = 0
promoted_to_verified = 0
for log in logs:
severity = str(log.get("severity", "WARNING")).upper()
if severity in ["INFO", "DEBUG"]:
continue
signature = log.get("signature", "unknown")
server = log.get("server", client_server)
message = log.get("message", "")
os_type = log.get("os_type", "unknown")
fp = f"{site_name}:{server}:{signature}"
cursor.execute(
"SELECT run_count, first_seen, last_seen, status, last_run_id FROM active_issues WHERE fingerprint = ?",
(fp,)
)
row = cursor.fetchone()
if row:
run_count, first_seen_str, last_seen_str, current_status, last_run_id = row
try:
last_seen_dt = datetime.fromisoformat(last_seen_str)
except Exception:
last_seen_dt = now
if (now - last_seen_dt) > timedelta(hours=window_hours):
new_runs = 1
new_first_seen = now_iso
new_status = "TRANSIENT"
else:
if last_run_id != run_id:
new_runs = run_count + 1
else:
new_runs = run_count
new_first_seen = first_seen_str
new_status = "VERIFIED" if new_runs >= min_runs else "TRANSIENT"
if new_status == "VERIFIED" and current_status != "VERIFIED":
promoted_to_verified += 1
cursor.execute("""
UPDATE active_issues
SET run_count = ?, last_seen = ?, first_seen = ?, status = ?, last_run_id = ?, message = ?, severity = ?
WHERE fingerprint = ?
""", (new_runs, now_iso, new_first_seen, new_status, run_id, message, severity, fp))
else:
initial_status = "VERIFIED" if 1 >= min_runs else "TRANSIENT"
cursor.execute("""
INSERT INTO active_issues
(fingerprint, site_name, server, signature, severity, message, os_type, first_seen, last_seen, run_count, status, last_run_id)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""", (fp, site_name, server, signature, severity, message, os_type, now_iso, now_iso, 1, initial_status, run_id))
processed_count += 1
conn.commit()
conn.close()
return {
"status": "success",
"run_id": run_id,
"processed": processed_count,
"promoted_verified": promoted_to_verified
}
async def handle_socket_client(reader: asyncio.StreamReader, writer: asyncio.StreamWriter):
try:
length_bytes = await reader.readexactly(4)
length = struct.unpack(">I", length_bytes)[0]
if length <= 0 or length > 10 * 1024 * 1024:
raise ValueError(f"Invalid frame size: {length}")
payload_bytes = await reader.readexactly(length)
envelope = json.loads(payload_bytes.decode("utf-8"))
expected_token = SERVER_STATE["config"]["auth_token"]
provided_token = envelope.get("auth_token")
if not secrets.compare_digest(str(provided_token), str(expected_token)):
err_msg = json.dumps({"status": "error", "message": "Authentication failed"}).encode("utf-8")
writer.write(struct.pack(">I", len(err_msg)) + err_msg)
await writer.drain()
writer.close()
await writer.wait_closed()
return
encrypted_armored = envelope.get("encrypted_payload", "")
pgp_msg = pgpy.PGPMessage.from_blob(encrypted_armored)
priv_key = SERVER_STATE["private_key_obj"]
decrypted_obj = priv_key.decrypt(pgp_msg)
decrypted_json_str = decrypted_obj.message
log_payload = json.loads(decrypted_json_str)
res = process_ingested_logs(
log_payload,
db_path=SERVER_STATE["config"]["db_path"],
window_hours=SERVER_STATE["config"]["evaluation_window_hours"],
min_runs=SERVER_STATE["config"]["min_persistence_runs"]
)
resp_bytes = json.dumps(res).encode("utf-8")
writer.write(struct.pack(">I", len(resp_bytes)) + resp_bytes)
await writer.drain()
except Exception as e:
err = json.dumps({"status": "error", "message": str(e)}).encode("utf-8")
try:
writer.write(struct.pack(">I", len(err)) + err)
await writer.drain()
except Exception:
pass
finally:
writer.close()
try:
await writer.wait_closed()
except Exception:
pass
@app.get("/api/hermes/report")
def get_hermes_report():
db_path = SERVER_STATE["config"]["db_path"]
window_hours = SERVER_STATE["config"]["evaluation_window_hours"]
min_runs = SERVER_STATE["config"]["min_persistence_runs"]
now = datetime.now(timezone.utc)
conn = sqlite3.connect(db_path)
cursor = conn.cursor()
cursor.execute("""
SELECT fingerprint, site_name, server, signature, severity, message, os_type, first_seen, last_seen, run_count, status
FROM active_issues
WHERE status = 'VERIFIED' AND run_count >= ?
""", (min_runs,))
rows = cursor.fetchall()
conn.close()
report = []
for r in rows:
last_seen_dt = datetime.fromisoformat(r[8])
if (now - last_seen_dt) <= timedelta(hours=window_hours):
report.append({
"fingerprint": r[0],
"site": r[1],
"server": r[2],
"signature": r[3],
"severity": r[4],
"message": r[5],
"os_type": r[6],
"first_seen": r[7],
"last_seen": r[8],
"consecutive_runs": r[9],
"evaluation_window": f"{window_hours}h",
"verified": True,
"status": r[10]
})
return report
@app.get("/api/hermes/all")
def get_all_issues():
db_path = SERVER_STATE["config"]["db_path"]
conn = sqlite3.connect(db_path)
cursor = conn.cursor()
cursor.execute("""
SELECT fingerprint, site_name, server, signature, severity, message, os_type, first_seen, last_seen, run_count, status
FROM active_issues
""")
rows = cursor.fetchall()
conn.close()
return [
{
"fingerprint": r[0],
"site": r[1],
"server": r[2],
"signature": r[3],
"severity": r[4],
"message": r[5],
"os_type": r[6],
"first_seen": r[7],
"last_seen": r[8],
"run_count": r[9],
"status": r[10]
}
for r in rows
]
@app.get("/health")
def health_check():
return {
"status": "healthy",
"server_name": SERVER_STATE["config"]["server_name"],
"fingerprint": SERVER_STATE["config"]["server_fingerprint"],
"tcp_port": SERVER_STATE["config"]["tcp_port"],
"hermes_port": SERVER_STATE["config"]["hermes_port"]
}
async def run_server():
config = SERVER_STATE["config"]
tcp_host = config["tcp_host"]
tcp_port = int(config["tcp_port"])
hermes_host = config["hermes_host"]
hermes_port = int(config["hermes_port"])
tcp_server = await asyncio.start_server(handle_socket_client, tcp_host, tcp_port)
print(f"[*] LOGAR TCP Socket Server listening on {tcp_host}:{tcp_port}")
uv_config = uvicorn.Config(app, host=hermes_host, port=hermes_port, log_level="warning")
uv_server = uvicorn.Server(uv_config)
print(f"[*] Hermes Reporting API available at http://{hermes_host}:{hermes_port}/api/hermes/report")
await asyncio.gather(
tcp_server.serve_forever(),
uv_server.serve()
)
def main():
parser = argparse.ArgumentParser(description="LOGAR Cloud Hub & TCP Socket Ingestion Server")
parser.add_argument("--config", default=CONFIG_FILE_NAME, help="Path to server_config.json")
parser.add_argument("--create-client-config", action="store_true", help="Generate a client config with encryption-only fingerprint and server address")
parser.add_argument("--client-out", default="client_config.json", help="Output file path for generated client config")
parser.add_argument("--server-host", default="127.0.0.1", help="Server address to embed in client config")
parser.add_argument("--server-port", type=int, default=None, help="TCP port to embed in client config")
args = parser.parse_args()
config = load_or_init_config(args.config)
init_db(config["db_path"])
priv_key_obj, _ = pgpy.PGPKey.from_blob(config["private_key"])
SERVER_STATE["config"] = config
SERVER_STATE["private_key_obj"] = priv_key_obj
if args.create_client_config:
port = args.server_port or config["tcp_port"]
create_client_config(
server_host=args.server_host,
server_port=port,
output_path=args.client_out,
config_path=args.config
)
sys.exit(0)
print("=" * 60)
print(f" LOGAR Server Hub: {config['server_name']}")
print(f" Encryption Fingerprint: {config['server_fingerprint']}")
print(f" Evaluation Window: {config['evaluation_window_hours']} hours | Rule: {config['min_persistence_runs']}+ consecutive runs")
print("=" * 60)
try:
asyncio.run(run_server())
except KeyboardInterrupt:
print("\n[!] Server shutting down.")
if __name__ == "__main__":
main()
-6
View File
@@ -1,6 +0,0 @@
pgpy>=0.6.0
standard-imghdr>=3.13.0; python_version >= "3.13"
cryptography>=42.0.0
fastapi>=0.110.0
uvicorn>=0.28.0
pydantic>=2.6.0
+115 -19
View File
@@ -1,31 +1,127 @@
# LOGAR Windows Edge Forwarder
Lightweight edge log forwarder for Windows servers.
Standalone compiled executable distribution for Windows Server and workstation environments.
## Features
- **Zero Local State**: No local database or state tracking. Forwarder simply scans recent logs and streams candidates.
- **Edge Noise Stripping**: Strips conversational/informational noise (INFO, DEBUG, Audit) at the source.
- **End-to-End OpenPGP Encryption**: Encrypts logs using the server's public key so that only the server can decrypt them.
- **Authenticated TCP Socket**: Connects directly via raw TCP framing with token verification.
- **No GPG Binary Required**: Pure-Python cryptography (`pgpy` + `cryptography`).
---
## Installation
```powershell
python -m pip install -r requirements.txt
## Overview
`Win_Client.exe` is a self-contained, pre-compiled executable that queries the Windows Application Event Log, filters candidate events at the source, encrypts the payload using OpenPGP, and streams records over an authenticated TCP socket to the central LOGAR hub.
### Key Capabilities
- **Pre-compiled & Dependency-Free**: Ships as a standalone native Windows executable (`Win_Client.exe`). No Python installation, pip packages, or GnuPG binaries are required on the host.
- **Source-Level Filtering**: Retains events spanning `INFO`, `WARNING`, and `ERROR`. Strips audit success/failure events and debug noise, skipping events older than 24 hours.
- **State Tracking & Deduplication**: Maintains persistent client state in `client_state.json` (tracking event record numbers and timestamp signatures) so every log record is forwarded exactly once without duplicates.
- **Fail-Safe State Commit**: State is committed only when the server returns a verified `success` response. In the event of a network outage, state remains unchanged and unsent events are retried automatically on the next run.
- **End-to-End Encryption**: Encrypts payloads using the server's OpenPGP public key before transmission.
---
## 1. Generating & Deploying the Configuration File
### Step 1: Generate `client_config.json` on the Server
Run the following command on your central LOGAR server to export a client bundle tailored for your environment:
```bash
python Server.py --create-client-config --server-host <SERVER_IP_OR_DNS> --server-port 9443 --client-out client_config.json
```
## Configuration
Place the `client_config.json` generated by the server (`Server.py --create-client-config`) in the same directory as `Win_Client.py`.
- Replace `<SERVER_IP_OR_DNS>` with the reachable IP address or FQDN of your central LOGAR server hub.
- Default TCP port is `9443`.
## Running the Forwarder
```powershell
python Win_Client.py --hours 6
### Step 2: Configuration Structure
The generated `client_config.json` contains:
```json
{
"server_host": "192.168.1.100",
"server_port": 9443,
"server_fingerprint": "375388960531264EA0648EC0D2C4E4ABC6F22AC2",
"server_public_key": "-----BEGIN PGP PUBLIC KEY BLOCK-----\n...",
"auth_token": "a1b2c3d4e5f6..."
}
```
## Scheduled Task Deployment
To run periodically via Windows Task Scheduler (e.g., every 3 hours):
> [!NOTE]
> A reference example is provided in `client_config.sample.json`. The configuration file contains **no host-specific names or site names** to ensure client anonymity and easy redistribution.
### Step 3: Copy to Edge Node
Place `Win_Client.exe` and `client_config.json` in the target directory (recommended: `C:\LOGAR\`):
```powershell
$Action = New-ScheduledTaskAction -Execute "python.exe" -Argument "C:\LOGAR\Win_Client.py --hours 6" -WorkingDirectory "C:\LOGAR"
New-Item -ItemType Directory -Path "C:\LOGAR" -Force
Copy-Item "Win_Client.exe", "client_config.json" -Destination "C:\LOGAR\"
```
---
## 2. Running Manually
Test the forwarder interactively from PowerShell or Command Prompt:
```powershell
cd C:\LOGAR
.\Win_Client.exe --hours 24
```
### Command-Line Arguments
| Argument | Default | Description |
| :--- | :--- | :--- |
| `--config` | `client_config.json` | Path to client configuration file |
| `--hours` | `24` | Lookback window in hours for event logs |
| `--state-file` | `client_state.json` | Path to persistent state tracking file |
| `--no-state` | `False` | Disable state tracking and send all events matching lookback window |
---
## 3. Installing as a Background Service / Scheduled Task
Edge forwarders run as episodic background processes (run, forward unsent candidate records, commit state, and terminate). On Windows, this is natively managed via Windows Task Scheduler running as a background service under `SYSTEM`.
### Method A: Windows Scheduled Task via PowerShell (Recommended)
Open an **Elevated PowerShell (Run as Administrator)** window and execute:
```powershell
# Define action and periodic trigger (every 3 hours indefinitely)
$Action = New-ScheduledTaskAction -Execute "C:\LOGAR\Win_Client.exe" -Argument "--hours 24" -WorkingDirectory "C:\LOGAR"
$Trigger = New-ScheduledTaskTrigger -Once -At (Get-Date) -RepetitionInterval (New-TimeSpan -Hours 3)
Register-ScheduledTask -TaskName "LOGAR_Windows_Forwarder" -Action $Action -Trigger $Trigger -Description "LOGAR Edge Forwarder"
# Configure task settings (wake on sleep, start when ready, run hidden)
$Settings = New-ScheduledTaskSettingsSet -AllowStartIfOnBatteries -DontStopIfGoingOnBatteries -StartWhenAvailable -ExecutionTimeLimit (New-TimeSpan -Minutes 15)
# Register task running under the local SYSTEM account with highest privileges
Register-ScheduledTask -TaskName "LOGAR_Forwarder" `
-Action $Action `
-Trigger $Trigger `
-Settings $Settings `
-User "NT AUTHORITY\SYSTEM" `
-RunLevel Highest `
-Description "LOGAR Windows Edge Log Forwarder Service"
# Verify task creation and trigger immediate execution
Start-ScheduledTask -TaskName "LOGAR_Forwarder"
Get-ScheduledTask -TaskName "LOGAR_Forwarder"
```
### Method B: Continuous Windows Service via NSSM
If your organizational policy requires a formal Windows Service listed under `services.msc`:
1. Download [NSSM (Non-Sucking Service Manager)](https://nssm.cc/).
2. Install the service using NSSM:
```cmd
nssm.exe install LOGAR_Forwarder "C:\LOGAR\Win_Client.exe" "--hours 24"
nssm.exe set LOGAR_Forwarder AppDirectory "C:\LOGAR"
nssm.exe set LOGAR_Forwarder AppRestartDelay 10800000
nssm.exe start LOGAR_Forwarder
```
*(Note: `AppRestartDelay 10800000` pauses 3 hours between execution cycles).*
---
## 4. Uninstallation & Removal
To remove the scheduled task:
```powershell
Unregister-ScheduledTask -TaskName "LOGAR_Forwarder" -Confirm:$false
Remove-Item -Recurse -Force "C:\LOGAR"
```
-211
View File
@@ -1,211 +0,0 @@
import os
import sys
import json
import socket
import struct
import argparse
import warnings
from datetime import datetime, timezone, timedelta
# Suppress cryptography / pgpy deprecation notices
warnings.filterwarnings("ignore")
import pgpy
try:
import win32evtlog
except ImportError:
win32evtlog = None
CONFIG_FILE_NAME = "client_config.json"
def load_config(config_path: str = CONFIG_FILE_NAME):
if not os.path.exists(config_path):
raise FileNotFoundError(
f"Client configuration file not found at: {config_path}\n"
f"Generate one from the server using: python Server.py --create-client-config --client-out {config_path}"
)
with open(config_path, "r", encoding="utf-8") as f:
return json.load(f)
def get_machine_identifier() -> str:
"""
Returns the hostname of the machine sending the logs,
and appends the network/DNS domain if available.
"""
fqdn = socket.getfqdn()
if fqdn and "." in fqdn and not fqdn.startswith("localhost"):
return fqdn
hostname = socket.gethostname()
user_dns_domain = os.environ.get("USERDNSDOMAIN")
if user_dns_domain and user_dns_domain.lower() != hostname.lower():
return f"{hostname}.{user_dns_domain.lower()}"
try:
host_ip = socket.gethostbyname(hostname)
canonical_name = socket.gethostbyaddr(host_ip)[0]
if canonical_name and "." in canonical_name and not canonical_name.startswith("localhost"):
return canonical_name
except Exception:
pass
return hostname
def get_recent_windows_logs(hours: int = 6):
"""
Scans the Windows Application Event Log backwards for events within the window.
Edge Thinness & Noise Stripping: INFO and DEBUG events are dropped at the source.
"""
if win32evtlog is None:
print("[!] pywin32 is not installed or not running on Windows. Returning mock/empty candidate list.")
return []
server = "localhost"
log_type = "Application"
flags = win32evtlog.EVENTLOG_BACKWARDS_READ | win32evtlog.EVENTLOG_SEQUENTIAL_READ
try:
hand = win32evtlog.OpenEventLog(server, log_type)
except Exception as e:
print(f"[!] Error opening Windows event log: {e}")
return []
logs = []
cutoff_time = datetime.now() - timedelta(hours=hours)
machine_id = get_machine_identifier()
sev_map = {
1: "CRITICAL",
2: "ERROR",
3: "WARNING"
}
while True:
events = win32evtlog.ReadEventLog(hand, flags, 0)
if not events:
break
for event in events:
if event.TimeGenerated < cutoff_time:
break
# Drop conversational or informational noise (INFO=4, etc.) at source
# Only retain Critical (1), Error (2), and Warning (3)
if event.EventType in sev_map:
msg = " ".join(event.StringInserts) if event.StringInserts else "Event Log Entry"
logs.append({
"server": machine_id,
"os_type": "windows",
"signature": event.SourceName or "Windows-Event",
"severity": sev_map[event.EventType],
"message": msg[:2048] # Limit message length
})
if events[-1].TimeGenerated < cutoff_time:
break
win32evtlog.CloseEventLog(hand)
return logs
def send_encrypted_logs_over_socket(config: dict, logs: list):
"""
Encrypts the payload using the server's OpenPGP public key and streams
over an authenticated TCP socket. Zero local state is maintained on the client.
"""
server_host = config["server_host"]
server_port = int(config["server_port"])
auth_token = config["auth_token"]
pub_key_armored = config["server_public_key"]
expected_fp = config.get("server_fingerprint", "").replace(" ", "").upper()
# Load and verify server public key
pub_key, _ = pgpy.PGPKey.from_blob(pub_key_armored)
actual_fp = str(pub_key.fingerprint).replace(" ", "").upper()
if expected_fp and actual_fp != expected_fp:
raise ValueError(
f"Server fingerprint mismatch! Expected {expected_fp}, but key has {actual_fp}."
)
machine_id = get_machine_identifier()
# Prepare zero-state candidate batch
payload = {
"server": machine_id,
"timestamp": datetime.now(timezone.utc).isoformat(),
"logs": logs
}
payload_json = json.dumps(payload)
# Encrypt payload with server's encryption-only key
pgp_msg = pgpy.PGPMessage.new(payload_json)
encrypted_msg = pub_key.encrypt(pgp_msg)
encrypted_armored = str(encrypted_msg)
# Envelope with socket authentication header
envelope = {
"auth_token": auth_token,
"timestamp": datetime.now(timezone.utc).isoformat(),
"encrypted_payload": encrypted_armored
}
envelope_bytes = json.dumps(envelope).encode("utf-8")
# Connect over TCP socket and transmit with 4-byte length prefix framing
print(f"[*] Connecting to LOGAR server at {server_host}:{server_port} over secure TCP socket...")
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
sock.settimeout(15.0)
sock.connect((server_host, server_port))
# Send frame: length (4 bytes big-endian) + envelope
frame = struct.pack(">I", len(envelope_bytes)) + envelope_bytes
sock.sendall(frame)
# Receive response length
resp_len_bytes = sock.recv(4)
if not resp_len_bytes:
raise ConnectionError("Server closed connection without response.")
resp_len = struct.unpack(">I", resp_len_bytes)[0]
resp_bytes = bytearray()
while len(resp_bytes) < resp_len:
chunk = sock.recv(min(4096, resp_len - len(resp_bytes)))
if not chunk:
break
resp_bytes.extend(chunk)
response = json.loads(resp_bytes.decode("utf-8"))
print(f"[+] Server response: {response}")
return response
def main():
parser = argparse.ArgumentParser(description="LOGAR Windows Edge Log Forwarder (Zero State)")
parser.add_argument("--config", default=CONFIG_FILE_NAME, help="Path to client_config.json")
parser.add_argument("--hours", type=int, default=6, help="Lookback window in hours for event logs")
args = parser.parse_args()
try:
config = load_config(args.config)
except Exception as e:
print(f"[!] Configuration error: {e}")
sys.exit(1)
machine_id = get_machine_identifier()
print(f"[*] Edge Forwarder Node: {machine_id}")
print(f"[*] Scanning Windows Application event log for candidate anomalies (last {args.hours} hours)...")
candidate_logs = get_recent_windows_logs(hours=args.hours)
print(f"[*] Found {len(candidate_logs)} candidate anomalies (noise stripped at source).")
try:
send_encrypted_logs_over_socket(config, candidate_logs)
except Exception as e:
print(f"[!] Failed to stream logs to server: {e}")
sys.exit(1)
if __name__ == "__main__":
main()
+1 -2
View File
@@ -3,6 +3,5 @@
"server_port": 9443,
"server_fingerprint": "PASTE_SERVER_FINGERPRINT_HERE",
"server_public_key": "-----BEGIN PGP PUBLIC KEY BLOCK-----\n...\n-----END PGP PUBLIC KEY BLOCK-----\n",
"auth_token": "PASTE_AUTH_TOKEN_HERE",
"site_name": "Frankfurt-DC"
"auth_token": "PASTE_AUTH_TOKEN_HERE"
}
-4
View File
@@ -1,4 +0,0 @@
pgpy>=0.6.0
standard-imghdr>=3.13.0; python_version >= "3.13"
cryptography>=42.0.0
pywin32>=306
-95
View File
@@ -1,95 +0,0 @@
import os
import sys
import json
import struct
import unittest
import warnings
warnings.filterwarnings("ignore")
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
import Win_Client
import pgpy
from pgpy.constants import PubKeyAlgorithm, KeyFlags, HashAlgorithm, SymmetricKeyAlgorithm, CompressionAlgorithm
class TestWinClientComponent(unittest.TestCase):
def setUp(self):
self.dummy_config = "test_win_client_config.json"
# Generate dummy PGP key for testing
key = pgpy.PGPKey.new(PubKeyAlgorithm.RSAEncryptOrSign, 2048)
uid = pgpy.PGPUID.new("TestHub")
key.add_uid(
uid,
usage={KeyFlags.EncryptCommunications, KeyFlags.EncryptStorage},
hashes=[HashAlgorithm.SHA256],
ciphers=[SymmetricKeyAlgorithm.AES256],
compression=[CompressionAlgorithm.Uncompressed]
)
self.server_priv = key
self.server_pub = key.pubkey
self.fingerprint = str(key.pubkey.fingerprint)
with open(self.dummy_config, "w", encoding="utf-8") as f:
json.dump({
"server_host": "127.0.0.1",
"server_port": 9443,
"server_fingerprint": self.fingerprint,
"server_public_key": str(self.server_pub),
"auth_token": "secret-test-token"
}, f)
def tearDown(self):
if os.path.exists(self.dummy_config):
try:
os.remove(self.dummy_config)
except Exception:
pass
def test_client_config_anonymity(self):
config = Win_Client.load_config(self.dummy_config)
self.assertNotIn("server_name", config)
self.assertNotIn("name", config)
self.assertNotIn("site_name", config)
self.assertEqual(config["server_fingerprint"], self.fingerprint)
def test_get_machine_identifier(self):
machine_id = Win_Client.get_machine_identifier()
self.assertIsInstance(machine_id, str)
self.assertGreater(len(machine_id), 0)
self.assertNotEqual(machine_id, "localhost")
def test_encryption_and_envelope_creation(self):
config = Win_Client.load_config(self.dummy_config)
logs = [{
"server": Win_Client.get_machine_identifier(),
"signature": "TestWinSignature",
"severity": "WARNING",
"message": "Disk space threshold warning"
}]
pub_key, _ = pgpy.PGPKey.from_blob(config["server_public_key"])
payload = {
"server": Win_Client.get_machine_identifier(),
"logs": logs
}
msg = pgpy.PGPMessage.new(json.dumps(payload))
enc = pub_key.encrypt(msg)
self.assertTrue(str(enc).startswith("-----BEGIN PGP MESSAGE-----"))
# Decrypt with private key to verify end-to-end payload integrity
dec = self.server_priv.decrypt(enc)
restored = json.loads(dec.message)
self.assertEqual(restored["logs"][0]["signature"], "TestWinSignature")
def test_framing_protocol(self):
envelope_data = json.dumps({"test": "data"}).encode("utf-8")
frame = struct.pack(">I", len(envelope_data)) + envelope_data
self.assertEqual(len(frame), 4 + len(envelope_data))
length = struct.unpack(">I", frame[:4])[0]
self.assertEqual(length, len(envelope_data))
if __name__ == "__main__":
unittest.main()
+141 -58
View File
@@ -5,9 +5,10 @@ import zipapp
import hashlib
import platform
import subprocess
import argparse
ROOT_DIR = os.path.abspath(os.path.dirname(__file__))
DIST_DIR = os.path.abspath("dist")
OUT_DIR = os.path.abspath("out")
BUILD_TEMP = os.path.abspath("build_temp")
def clean_and_prep():
@@ -36,84 +37,166 @@ def build_pyinstaller_binary(script_path, binary_name):
raise RuntimeError(f"Failed to build {binary_name}")
print(f"[+] Successfully compiled {binary_name}")
def build_linux_zipapp_binary():
print("[*] Packaging Linux_Client.bin executable binary...")
def calculate_sha256(filepath):
h = hashlib.sha256()
with open(filepath, "rb") as f:
while chunk := f.read(65536):
h.update(chunk)
return h.hexdigest()
def write_checksum_file(filename, digest, binary_filename):
out_path = os.path.join(DIST_DIR, filename)
with open(out_path, "w", encoding="utf-8") as f:
f.write(f"{digest} {binary_filename}\n")
print(f"[+] Generated checksum file: {filename} ({digest[:16]}...)")
def build_linux_zipapp_fallback():
print("[*] Packaging Linux standalone zipapp fallback binaries...")
# Linux Client zipapp
app_dir = os.path.join(BUILD_TEMP, "linux_app")
os.makedirs(app_dir, exist_ok=True)
shutil.copy(os.path.join(OUT_DIR, "linux_client", "Linux_Client.py"), os.path.join(app_dir, "Linux_Client.py"))
bin_output = os.path.join(DIST_DIR, "Linux_Client.bin")
shutil.copy(os.path.join(ROOT_DIR, "Linux_Client.py"), os.path.join(app_dir, "Linux_Client.py"))
client_out = os.path.join(DIST_DIR, "Linux_Client.bin")
zipapp.create_archive(
source=app_dir,
target=bin_output,
target=client_out,
interpreter="/usr/bin/env python3",
main="Linux_Client:main"
)
print(f"[+] Successfully generated {bin_output}")
# Server zipapp
srv_dir = os.path.join(BUILD_TEMP, "linux_srv")
os.makedirs(srv_dir, exist_ok=True)
shutil.copy(os.path.join(ROOT_DIR, "Server.py"), os.path.join(srv_dir, "Server.py"))
server_out = os.path.join(DIST_DIR, "Server.bin")
zipapp.create_archive(
source=srv_dir,
target=server_out,
interpreter="/usr/bin/env python3",
main="Server:main"
)
def generate_checksums():
checksum_file = os.path.join(DIST_DIR, "SHA256SUMS.txt")
lines = []
for fname in sorted(os.listdir(DIST_DIR)):
if fname == "SHA256SUMS.txt":
continue
fpath = os.path.join(DIST_DIR, fname)
if os.path.isfile(fpath):
with open(fpath, "rb") as f:
digest = hashlib.sha256(f.read()).hexdigest()
lines.append(f"{digest} {fname}")
with open(checksum_file, "w", encoding="utf-8") as f:
f.write("\n".join(lines) + "\n")
def build_windows():
print("[*] Compiling Windows standalone executables...")
win_client_script = os.path.join(ROOT_DIR, "Win_Client.py")
build_pyinstaller_binary(win_client_script, "Win_Client")
server_script = os.path.join(ROOT_DIR, "Server.py")
build_pyinstaller_binary(server_script, "Server")
client_bin = os.path.join(DIST_DIR, "Win_Client.exe")
server_bin = os.path.join(DIST_DIR, "Server.exe")
sums = []
if os.path.exists(client_bin):
client_hash = calculate_sha256(client_bin)
write_checksum_file("win_client_sha256sum", client_hash, "Win_Client.exe")
write_checksum_file("win_agent_sha256sum", client_hash, "Win_Client.exe")
sums.append(f"{client_hash} Win_Client.exe")
else:
print(f"[!] Warning: Expected {client_bin} was not found.")
if os.path.exists(server_bin):
server_hash = calculate_sha256(server_bin)
write_checksum_file("win_server_sha256sum", server_hash, "Server.exe")
sums.append(f"{server_hash} Server.exe")
else:
print(f"[!] Warning: Expected {server_bin} was not found.")
sums_path = os.path.join(DIST_DIR, "SHA256SUMS_windows.txt")
with open(sums_path, "w", encoding="utf-8") as f:
f.write("\n".join(sums) + "\n")
def build_linux():
print("[*] Compiling Linux standalone binaries...")
is_linux_host = platform.system() == "Linux"
if is_linux_host:
linux_client_script = os.path.join(ROOT_DIR, "Linux_Client.py")
build_pyinstaller_binary(linux_client_script, "Linux_Client.bin")
server_script = os.path.join(ROOT_DIR, "Server.py")
build_pyinstaller_binary(server_script, "Server.bin")
# Normalize extensions in case PyInstaller dropped .bin
for name in ["Linux_Client", "Server"]:
plain_path = os.path.join(DIST_DIR, name)
bin_path = os.path.join(DIST_DIR, f"{name}.bin")
if os.path.exists(plain_path) and not os.path.exists(bin_path):
os.rename(plain_path, bin_path)
# Ensure executable permissions on Linux
for b in ["Linux_Client.bin", "Server.bin"]:
p = os.path.join(DIST_DIR, b)
if os.path.exists(p):
try:
os.chmod(p, 0o755)
except Exception:
pass
else:
print("[!] Note: Host platform is not Linux. Generating executable zipapps for Linux target.")
build_linux_zipapp_fallback()
client_bin = os.path.join(DIST_DIR, "Linux_Client.bin")
server_bin = os.path.join(DIST_DIR, "Server.bin")
sums = []
if os.path.exists(client_bin):
client_hash = calculate_sha256(client_bin)
write_checksum_file("linux_client_sha256sum", client_hash, "Linux_Client.bin")
write_checksum_file("linux_agent_sha256sum", client_hash, "Linux_Client.bin")
sums.append(f"{client_hash} Linux_Client.bin")
else:
print(f"[!] Warning: Expected {client_bin} was not found.")
if os.path.exists(server_bin):
server_hash = calculate_sha256(server_bin)
write_checksum_file("linux_server_sha256sum", server_hash, "Server.bin")
sums.append(f"{server_hash} Server.bin")
else:
print(f"[!] Warning: Expected {server_bin} was not found.")
sums_path = os.path.join(DIST_DIR, "SHA256SUMS_linux.txt")
with open(sums_path, "w", encoding="utf-8") as f:
f.write("\n".join(sums) + "\n")
def main(target=None):
if target is None:
parser = argparse.ArgumentParser(description="LOGAR Standalone Binary Compiler & Packager")
parser.add_argument(
"--target", "-t",
choices=["windows", "win", "linux", "auto"],
default="auto",
help="Target platform to compile binaries for (default: auto-detect)"
)
args, _ = parser.parse_known_args()
target = args.target
if target == "auto":
target = "windows" if platform.system() == "Windows" else "linux"
elif target == "win":
target = "windows"
def main():
print("=" * 60)
print(" LOGAR Binary Packaging (Binaries Only)")
print(f" Platform: {platform.system()} ({platform.machine()})")
print(" LOGAR Binary Packaging")
print(f" Host Platform: {platform.system()} ({platform.machine()})")
print(f" Target Platform: {target.upper()}")
print("=" * 60)
clean_and_prep()
is_windows = platform.system() == "Windows"
if is_windows:
# Build Windows client executable
win_client_script = os.path.join(OUT_DIR, "win_client", "Win_Client.py")
build_pyinstaller_binary(win_client_script, "Win_Client")
# Build Windows server executable
server_script = os.path.join(OUT_DIR, "server", "Server.py")
build_pyinstaller_binary(server_script, "Server")
# Build Linux client standalone binary
build_linux_zipapp_binary()
if target == "windows":
build_windows()
elif target == "linux":
build_linux()
else:
# On Linux runner: Build native Linux binaries
linux_client_script = os.path.join(OUT_DIR, "linux_client", "Linux_Client.py")
build_pyinstaller_binary(linux_client_script, "Linux_Client.bin")
server_script = os.path.join(OUT_DIR, "server", "Server.py")
build_pyinstaller_binary(server_script, "Server.bin")
# Also package standalone Windows zipapp executable
win_app_dir = os.path.join(BUILD_TEMP, "win_app")
os.makedirs(win_app_dir, exist_ok=True)
shutil.copy(os.path.join(OUT_DIR, "win_client", "Win_Client.py"), os.path.join(win_app_dir, "Win_Client.py"))
win_bin_output = os.path.join(DIST_DIR, "Win_Client.pyz")
zipapp.create_archive(
source=win_app_dir,
target=win_bin_output,
interpreter="/usr/bin/env python3",
main="Win_Client:main"
)
generate_checksums()
raise ValueError(f"Unsupported target: {target}")
# Clean temporary build directory
if os.path.exists(BUILD_TEMP):
shutil.rmtree(BUILD_TEMP, ignore_errors=True)
print("\n[+] Binary shipping artifacts assembled in 'dist/':")
for f in os.listdir(DIST_DIR):
for f in sorted(os.listdir(DIST_DIR)):
sz = os.path.getsize(os.path.join(DIST_DIR, f))
print(f" - {f} ({sz / (1024*1024):.2f} MB)" if sz > 1024*1024 else f" - {f} ({sz} bytes)")
print("=" * 60)
+6 -1
View File
@@ -120,9 +120,14 @@ def run_tests():
print("\n=== [5] Testing Windows Client Script Integration ===")
from Win_Client import get_recent_windows_logs
win_logs = get_recent_windows_logs(hours=6)
win_logs = get_recent_windows_logs(hours=24)
print(f"[Win_Client] Successfully queried Windows logs: {len(win_logs)} candidate entries.")
print("\n=== [6] Testing Linux Client Script Integration ===")
from Linux_Client import get_recent_linux_logs
linux_logs = get_recent_linux_logs(hours=24)
print(f"[Linux_Client] Successfully queried Linux logs: {len(linux_logs)} candidate entries.")
print("\n==========================================")
print(" ALL VERIFICATION TESTS PASSED SUCCESSFULLY! ")
print("==========================================")
+202
View File
@@ -0,0 +1,202 @@
import os
import sys
import json
import struct
import unittest
import warnings
warnings.filterwarnings("ignore")
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
import Linux_Client
import pgpy
from pgpy.constants import PubKeyAlgorithm, KeyFlags, HashAlgorithm, SymmetricKeyAlgorithm, CompressionAlgorithm
class TestLinuxClientComponent(unittest.TestCase):
def setUp(self):
self.dummy_config = "test_linux_client_config.json"
# Generate dummy PGP key for testing
key = pgpy.PGPKey.new(PubKeyAlgorithm.RSAEncryptOrSign, 2048)
uid = pgpy.PGPUID.new("TestHub")
key.add_uid(
uid,
usage={KeyFlags.EncryptCommunications, KeyFlags.EncryptStorage},
hashes=[HashAlgorithm.SHA256],
ciphers=[SymmetricKeyAlgorithm.AES256],
compression=[CompressionAlgorithm.Uncompressed]
)
self.server_priv = key
self.server_pub = key.pubkey
self.fingerprint = str(key.pubkey.fingerprint)
with open(self.dummy_config, "w", encoding="utf-8") as f:
json.dump({
"server_host": "127.0.0.1",
"server_port": 9443,
"server_fingerprint": self.fingerprint,
"server_public_key": str(self.server_pub),
"auth_token": "secret-test-token"
}, f)
def tearDown(self):
if os.path.exists(self.dummy_config):
try:
os.remove(self.dummy_config)
except Exception:
pass
def test_client_config_anonymity(self):
config = Linux_Client.load_config(self.dummy_config)
self.assertNotIn("server_name", config)
self.assertNotIn("name", config)
self.assertNotIn("site_name", config)
self.assertEqual(config["server_fingerprint"], self.fingerprint)
def test_get_machine_identifier(self):
machine_id = Linux_Client.get_machine_identifier()
self.assertIsInstance(machine_id, str)
self.assertGreater(len(machine_id), 0)
self.assertNotEqual(machine_id, "localhost")
def test_journalctl_parsing_and_priority_filter(self):
sample_journal_lines = [
json.dumps({"PRIORITY": "3", "SYSLOG_IDENTIFIER": "sshd", "MESSAGE": "Failed password for root"}),
json.dumps({"PRIORITY": "4", "SYSLOG_IDENTIFIER": "systemd", "MESSAGE": "Unit entered failed state"}),
json.dumps({"PRIORITY": "6", "SYSLOG_IDENTIFIER": "cron", "MESSAGE": "Informational session opened"}),
json.dumps({"PRIORITY": "7", "SYSLOG_IDENTIFIER": "debugd", "MESSAGE": "Verbose debugging log"}),
]
logs = []
machine_id = Linux_Client.get_machine_identifier()
for line_str in sample_journal_lines:
entry = json.loads(line_str)
priority = int(entry.get("PRIORITY", "6"))
# Filter: Retain INFO to ERROR (<= 6), drop DEBUG (> 6)
if priority > 6:
continue
if priority <= 3:
sev = "ERROR"
elif priority in (4, 5):
sev = "WARNING"
else:
sev = "INFO"
logs.append({
"server": machine_id,
"os_type": "linux",
"signature": entry.get("SYSLOG_IDENTIFIER", "unknown"),
"severity": sev,
"message": entry.get("MESSAGE", "")
})
# Priority 7 (DEBUG) must be stripped, while 3 (ERROR), 4 (WARNING), 6 (INFO) are retained
self.assertEqual(len(logs), 3)
self.assertEqual(logs[0]["severity"], "ERROR")
self.assertEqual(logs[1]["severity"], "WARNING")
self.assertEqual(logs[2]["severity"], "INFO")
def test_encryption_and_decryption(self):
config = Linux_Client.load_config(self.dummy_config)
pub_key, _ = pgpy.PGPKey.from_blob(config["server_public_key"])
payload = {
"server": Linux_Client.get_machine_identifier(),
"logs": [{"signature": "kernel", "severity": "ERROR", "message": "Kernel panic - not syncing"}]
}
msg = pgpy.PGPMessage.new(json.dumps(payload))
enc = pub_key.encrypt(msg)
self.assertTrue(str(enc).startswith("-----BEGIN PGP MESSAGE-----"))
dec = self.server_priv.decrypt(enc)
restored = json.loads(dec.message)
self.assertEqual(restored["logs"][0]["signature"], "kernel")
def test_state_lifecycle(self):
state_path = "test_linux_state.json"
try:
# 1. Load non-existent returns empty dict
state = Linux_Client.load_state(state_path)
self.assertEqual(state, {})
# 2. Stage new cursor and timestamp
state["new_last_cursor"] = "s=abc;i=123"
state["new_last_timestamp_us"] = 1700000000000000
state["new_sent_cursors"] = ["s=abc;i=123"]
# 3. Commit state moves staged keys to permanent and writes atomically
Linux_Client.commit_state(state, state_path)
self.assertNotIn("new_last_cursor", state)
self.assertEqual(state.get("last_cursor"), "s=abc;i=123")
self.assertEqual(state.get("last_timestamp_us"), 1700000000000000)
self.assertEqual(state.get("sent_cursors"), ["s=abc;i=123"])
# 4. Reload from disk
reloaded = Linux_Client.load_state(state_path)
self.assertEqual(reloaded.get("last_cursor"), "s=abc;i=123")
self.assertEqual(reloaded.get("last_timestamp_us"), 1700000000000000)
finally:
if os.path.exists(state_path):
os.remove(state_path)
def test_duplicate_suppression_and_lookback_logic(self):
from datetime import datetime, timezone, timedelta
now_us = datetime.now(timezone.utc).timestamp() * 1_000_000
cutoff_epoch_us = (datetime.now(timezone.utc) - timedelta(hours=24)).timestamp() * 1_000_000
mock_entries = [
# 1. 26 hours old -> skip (> 24h)
{"__CURSOR": "c1", "__REALTIME_TIMESTAMP": str(int(now_us - 26 * 3600 * 1_000_000)), "PRIORITY": "3", "MESSAGE": "Old error"},
# 2. 2 hours old, already sent -> skip
{"__CURSOR": "c2", "__REALTIME_TIMESTAMP": str(int(now_us - 2 * 3600 * 1_000_000)), "PRIORITY": "4", "MESSAGE": "Already sent warning"},
# 3. 1 hour old, new entry -> retain
{"__CURSOR": "c3", "__REALTIME_TIMESTAMP": str(int(now_us - 1 * 3600 * 1_000_000)), "PRIORITY": "6", "MESSAGE": "New info"},
# 4. 30 mins old, debug -> skip priority
{"__CURSOR": "c4", "__REALTIME_TIMESTAMP": str(int(now_us - 1800 * 1_000_000)), "PRIORITY": "7", "MESSAGE": "Debug entry"}
]
state = {
"last_cursor": "c2",
"last_timestamp_us": int(now_us - 2 * 3600 * 1_000_000),
"sent_cursors": ["c2"]
}
# Simulate the filtering loop from get_recent_linux_logs
logs = []
last_cursor = state.get("last_cursor")
last_timestamp_us = float(state.get("last_timestamp_us", 0))
sent_cursors = set(state.get("sent_cursors", []))
newest_cursor = None
newest_timestamp_us = last_timestamp_us
collected_cursors = []
for entry in mock_entries:
entry_cursor = entry.get("__CURSOR")
entry_ts_us = float(entry.get("__REALTIME_TIMESTAMP"))
if entry_ts_us < cutoff_epoch_us:
continue
if entry_cursor and (entry_cursor in sent_cursors or entry_cursor == last_cursor):
continue
if last_timestamp_us > 0 and entry_ts_us < last_timestamp_us:
continue
if entry_cursor:
newest_cursor = entry_cursor
collected_cursors.append(entry_cursor)
if entry_ts_us > newest_timestamp_us:
newest_timestamp_us = entry_ts_us
priority = int(entry.get("PRIORITY", "6"))
if priority > 6:
continue
logs.append(entry)
self.assertEqual(len(logs), 1)
self.assertEqual(logs[0]["__CURSOR"], "c3")
self.assertEqual(newest_cursor, "c4")
if __name__ == "__main__":
unittest.main()
@@ -114,6 +114,33 @@ class TestServerComponent(unittest.TestCase):
self.assertEqual(row[0], 4)
self.assertEqual(row[1], "VERIFIED")
def test_server_severity_filtering(self):
Server.init_db(self.test_db)
payload = {
"server": "app-worker-01.corp.local",
"logs": [
{"server": "app-worker-01", "signature": "SigInfo", "severity": "INFO", "message": "Info msg", "os_type": "linux"},
{"server": "app-worker-01", "signature": "SigWarn", "severity": "WARNING", "message": "Warn msg", "os_type": "linux"},
{"server": "app-worker-01", "signature": "SigErr", "severity": "ERROR", "message": "Err msg", "os_type": "linux"},
{"server": "app-worker-01", "signature": "SigDebug", "severity": "DEBUG", "message": "Debug msg", "os_type": "linux"},
{"server": "app-worker-01", "signature": "SigTrace", "severity": "TRACE", "message": "Trace msg", "os_type": "linux"}
]
}
res = Server.process_ingested_logs(payload, self.test_db, window_hours=12, min_runs=4)
self.assertEqual(res["status"], "success")
conn = sqlite3.connect(self.test_db)
c = conn.cursor()
c.execute("SELECT signature FROM active_issues ORDER BY signature")
sigs = [r[0] for r in c.fetchall()]
conn.close()
self.assertIn("SigInfo", sigs)
self.assertIn("SigWarn", sigs)
self.assertIn("SigErr", sigs)
self.assertNotIn("SigDebug", sigs)
self.assertNotIn("SigTrace", sigs)
if __name__ == "__main__":
unittest.main()
+188
View File
@@ -0,0 +1,188 @@
import os
import sys
import json
import struct
import unittest
import warnings
warnings.filterwarnings("ignore")
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
import Win_Client
import pgpy
from pgpy.constants import PubKeyAlgorithm, KeyFlags, HashAlgorithm, SymmetricKeyAlgorithm, CompressionAlgorithm
class TestWinClientComponent(unittest.TestCase):
def setUp(self):
self.dummy_config = "test_win_client_config.json"
# Generate dummy PGP key for testing
key = pgpy.PGPKey.new(PubKeyAlgorithm.RSAEncryptOrSign, 2048)
uid = pgpy.PGPUID.new("TestHub")
key.add_uid(
uid,
usage={KeyFlags.EncryptCommunications, KeyFlags.EncryptStorage},
hashes=[HashAlgorithm.SHA256],
ciphers=[SymmetricKeyAlgorithm.AES256],
compression=[CompressionAlgorithm.Uncompressed]
)
self.server_priv = key
self.server_pub = key.pubkey
self.fingerprint = str(key.pubkey.fingerprint)
with open(self.dummy_config, "w", encoding="utf-8") as f:
json.dump({
"server_host": "127.0.0.1",
"server_port": 9443,
"server_fingerprint": self.fingerprint,
"server_public_key": str(self.server_pub),
"auth_token": "secret-test-token"
}, f)
def tearDown(self):
if os.path.exists(self.dummy_config):
try:
os.remove(self.dummy_config)
except Exception:
pass
def test_client_config_anonymity(self):
config = Win_Client.load_config(self.dummy_config)
self.assertNotIn("server_name", config)
self.assertNotIn("name", config)
self.assertNotIn("site_name", config)
self.assertEqual(config["server_fingerprint"], self.fingerprint)
def test_get_machine_identifier(self):
machine_id = Win_Client.get_machine_identifier()
self.assertIsInstance(machine_id, str)
self.assertGreater(len(machine_id), 0)
self.assertNotEqual(machine_id, "localhost")
def test_encryption_and_envelope_creation(self):
config = Win_Client.load_config(self.dummy_config)
logs = [{
"server": Win_Client.get_machine_identifier(),
"signature": "TestWinSignature",
"severity": "WARNING",
"message": "Disk space threshold warning"
}]
pub_key, _ = pgpy.PGPKey.from_blob(config["server_public_key"])
payload = {
"server": Win_Client.get_machine_identifier(),
"logs": logs
}
msg = pgpy.PGPMessage.new(json.dumps(payload))
enc = pub_key.encrypt(msg)
self.assertTrue(str(enc).startswith("-----BEGIN PGP MESSAGE-----"))
# Decrypt with private key to verify end-to-end payload integrity
dec = self.server_priv.decrypt(enc)
restored = json.loads(dec.message)
self.assertEqual(restored["logs"][0]["signature"], "TestWinSignature")
def test_framing_protocol(self):
envelope_data = json.dumps({"test": "data"}).encode("utf-8")
frame = struct.pack(">I", len(envelope_data)) + envelope_data
self.assertEqual(len(frame), 4 + len(envelope_data))
length = struct.unpack(">I", frame[:4])[0]
self.assertEqual(length, len(envelope_data))
def test_windows_event_filtering_and_severity_map(self):
# sev_map: 1 -> ERROR, 2 -> WARNING, 4 -> INFO
sev_map = {1: "ERROR", 2: "WARNING", 4: "INFO"}
raw_event_types = [1, 2, 4, 8, 16] # 8 is Audit Success, 16 is Audit Failure
filtered = [sev_map[et] for et in raw_event_types if et in sev_map]
self.assertEqual(filtered, ["ERROR", "WARNING", "INFO"])
def test_state_lifecycle(self):
state_path = "test_win_state.json"
try:
# 1. Load non-existent returns empty dict
state = Win_Client.load_state(state_path)
self.assertEqual(state, {})
# 2. Stage new record number and sent IDs
state["new_last_record_number"] = 42
state["new_sent_record_ids"] = ["42:2026-09-04T12:00:00"]
# 3. Commit state moves staged keys to permanent and writes atomically
Win_Client.commit_state(state, state_path)
self.assertNotIn("new_last_record_number", state)
self.assertEqual(state.get("last_record_number"), 42)
self.assertEqual(state.get("sent_record_ids"), ["42:2026-09-04T12:00:00"])
# 4. Reload from disk
reloaded = Win_Client.load_state(state_path)
self.assertEqual(reloaded.get("last_record_number"), 42)
self.assertEqual(reloaded.get("sent_record_ids"), ["42:2026-09-04T12:00:00"])
finally:
if os.path.exists(state_path):
os.remove(state_path)
def test_duplicate_suppression_and_lookback_logic(self):
from datetime import datetime, timezone, timedelta
now = datetime.now()
cutoff_time = now - timedelta(hours=24)
# Mock event object
class MockEvent:
def __init__(self, rec_num, time_gen, event_type, source="TestApp", inserts=None):
self.RecordNumber = rec_num
self.TimeGenerated = time_gen
self.EventType = event_type
self.SourceName = source
self.StringInserts = inserts or ["Test"]
# Events read backwards: newest (rec 103) down to older (rec 99)
mock_events = [
# 1. New error within last 24h
MockEvent(103, now - timedelta(hours=1), 1),
# 2. New warning within last 24h
MockEvent(102, now - timedelta(hours=2), 2),
# 3. Already sent event (rec 101)
MockEvent(101, now - timedelta(hours=3), 4),
# 4. Event at or before last_record_number (rec 100) -> should stop backwards scan
MockEvent(100, now - timedelta(hours=4), 1),
# 5. Old event (> 24h)
MockEvent(99, now - timedelta(hours=26), 1),
]
state = {
"last_record_number": 100,
"sent_record_ids": ["101:" + (now - timedelta(hours=3)).isoformat()]
}
sev_map = {1: "ERROR", 2: "WARNING", 4: "INFO"}
logs = []
last_record_number = int(state.get("last_record_number", 0))
sent_record_ids = set(state.get("sent_record_ids", []))
newest_record_number = 0
for event in mock_events:
rec_num = int(event.RecordNumber)
if newest_record_number == 0:
newest_record_number = rec_num
if event.TimeGenerated < cutoff_time:
break
if last_record_number > 0 and newest_record_number >= last_record_number:
if rec_num <= last_record_number:
break
rec_id = f"{rec_num}:{event.TimeGenerated.isoformat()}"
if rec_id in sent_record_ids:
continue
if event.EventType in sev_map:
logs.append(rec_num)
# Only rec 103 and 102 should be processed (101 is already sent, <= 100 breaks early)
self.assertEqual(logs, [103, 102])
if __name__ == "__main__":
unittest.main()
+74 -5
View File
@@ -7,6 +7,8 @@ import urllib.parse
import mimetypes
import package_dist
import time
DEFAULT_GITEA_URL = os.environ.get("GITEA_SERVER_URL", "https://gitea.eibl.tech")
DEFAULT_REPO = os.environ.get("GITEA_REPOSITORY", "me0nline/LOGAR")
@@ -28,7 +30,7 @@ def delete_asset(base_url, repo, release_id, asset_id, token):
except Exception:
pass
def upload_file_to_release(base_url, repo, release_id, token, file_path):
def upload_file_to_release(base_url, repo, release_id, token, file_path, max_retries=3):
filename = os.path.basename(file_path)
# Clean up existing asset with same name if already present
@@ -42,6 +44,7 @@ def upload_file_to_release(base_url, repo, release_id, token, file_path):
with open(file_path, "rb") as f:
file_bytes = f.read()
for attempt in range(1, max_retries + 1):
req = urllib.request.Request(url, data=file_bytes, method="POST")
req.add_header("Authorization", f"token {token}")
req.add_header("Content-Type", "application/octet-stream")
@@ -54,7 +57,16 @@ def upload_file_to_release(base_url, repo, release_id, token, file_path):
return data
except urllib.error.HTTPError as e:
err = e.read().decode("utf-8", errors="ignore")
print(f"[!] Error uploading {filename}: HTTP {e.code} - {err}")
print(f"[!] Attempt {attempt}/{max_retries} - Error uploading {filename}: HTTP {e.code} - {err}")
if attempt < max_retries:
time.sleep(2 * attempt)
else:
return None
except Exception as ex:
print(f"[!] Attempt {attempt}/{max_retries} - Exception uploading {filename}: {ex}")
if attempt < max_retries:
time.sleep(2 * attempt)
else:
return None
def create_or_get_release(base_url, repo, tag, token, title=None, notes=None):
@@ -72,6 +84,24 @@ def create_or_get_release(base_url, repo, tag, token, title=None, notes=None):
with urllib.request.urlopen(check_req) as resp:
existing = json.loads(resp.read().decode("utf-8"))
print(f"[*] Found existing release for tag {tag} (ID: {existing['id']})")
if notes or title:
patch_url = f"{base_url}/api/v1/repos/{repo}/releases/{existing['id']}"
patch_payload = {}
if title:
patch_payload["name"] = title
if notes:
patch_payload["body"] = notes
patch_req = urllib.request.Request(
patch_url,
data=json.dumps(patch_payload).encode("utf-8"),
headers=headers,
method="PATCH"
)
try:
with urllib.request.urlopen(patch_req) as p_resp:
print(f"[+] Updated release description for tag {tag}")
except Exception as e:
print(f"[!] Warning: Could not update existing release description: {e}")
return existing["id"]
except urllib.error.HTTPError:
pass
@@ -85,23 +115,39 @@ def create_or_get_release(base_url, repo, tag, token, title=None, notes=None):
"prerelease": False
}
req = urllib.request.Request(url, data=json.dumps(payload).encode("utf-8"), headers=headers, method="POST")
try:
with urllib.request.urlopen(req) as resp:
created = json.loads(resp.read().decode("utf-8"))
print(f"[+] Created release {tag} (ID: {created['id']})")
return created["id"]
except urllib.error.HTTPError as e:
print(f"[*] Release creation returned HTTP {e.code}. Checking if peer runner created it concurrently...")
for attempt in range(1, 6):
time.sleep(2)
try:
with urllib.request.urlopen(check_req) as resp:
existing = json.loads(resp.read().decode("utf-8"))
print(f"[+] Retrieved peer-created release for tag {tag} (ID: {existing['id']})")
return existing["id"]
except Exception:
pass
raise
def main():
parser = argparse.ArgumentParser(description="Upload LOGAR compiled binaries directly to Gitea Release")
parser.add_argument("--tag", default=os.environ.get("GITEA_REF_NAME"), help="Release tag name (e.g. v1.0.0)")
parser.add_argument("--tag", default=os.environ.get("GITEA_REF_NAME"), help="Release tag name (e.g. v1.0.1)")
parser.add_argument("--token", default=os.environ.get("GITEA_TOKEN"), help="Gitea Personal Access Token (or set GITEA_TOKEN env var)")
parser.add_argument("--url", default=DEFAULT_GITEA_URL, help="Base Gitea instance URL")
parser.add_argument("--repo", default=DEFAULT_REPO, help="Repository owner/name")
parser.add_argument("--title", default=os.environ.get("RELEASE_TITLE"), help="Release title")
parser.add_argument("--notes", default=os.environ.get("RELEASE_NOTES"), help="Release description / notes")
parser.add_argument("--notes-file", default=None, help="Path to markdown file with release notes")
parser.add_argument("--skip-build", action="store_true", help="Skip running package_dist.py before upload")
args = parser.parse_args()
tag = args.tag
if not tag:
tag = input("Enter tag name (e.g. v1.0.0): ").strip()
tag = input("Enter tag name (e.g. v1.0.1): ").strip()
token = args.token
if not token:
@@ -111,6 +157,29 @@ def main():
print("[!] Tag and Token are required.")
sys.exit(1)
# Resolve release notes
notes = args.notes
if not notes and args.notes_file and os.path.exists(args.notes_file):
with open(args.notes_file, "r", encoding="utf-8") as nf:
notes = nf.read()
elif not notes and os.path.exists("RELEASE_NOTES.md"):
with open("RELEASE_NOTES.md", "r", encoding="utf-8") as nf:
notes = nf.read()
elif not notes:
notes = (
f"## LOGAR Release {tag}\n\n"
"### Changes in this Release:\n"
"- **Dual Platform Gitea Release Automation**: Added dedicated Windows (`release-windows.yml`) and Linux (`release-linux.yml`) Gitea Actions to compile native executables and publish assets concurrently.\n"
"- **Dedicated SHA-256 Checksums**: Release assets now include dedicated checksum files matching `[win/linux]_[client/agent]_sha256sum` (e.g., `win_client_sha256sum`, `win_agent_sha256sum`, `win_server_sha256sum`, `linux_client_sha256sum`, `linux_agent_sha256sum`, `linux_server_sha256sum`).\n"
"- **Removed Client Filter Logic**: Removed restrictive source-level noise filtering on edge forwarders. Clients now stream all candidate events from `INFO` up to `ERROR` across the lookback window instead of discarding them at the source.\n"
"- **State Tracking & Deduplication**: Added persistent state tracking (`client_state.json`) with cursor and record number deduplication so previously transmitted events are never resent.\n"
"- **24-Hour Lookback Window**: Forwarders now scan and upload events from the last 24 hours, skipping older entries.\n"
"- **Lightweight Distribution Structure**: Cleaned `out/` to strictly contain deployment documentation and sample configurations.\n"
"- **Automated Gitea CI/CD**: Integrated push testing workflow (`ci.yml`) and automated release asset packaging.\n"
)
title = args.title or f"LOGAR Release {tag}"
if not args.skip_build:
print("[*] Assembling compiled binaries...")
package_dist.main()
@@ -121,7 +190,7 @@ def main():
sys.exit(1)
print(f"[*] Connecting to Gitea: {args.url} (repo: {args.repo})...")
release_id = create_or_get_release(args.url, args.repo, tag, token)
release_id = create_or_get_release(args.url, args.repo, tag, token, title=title, notes=notes)
print(f"[*] Uploading binary assets from '{dist_dir}'...")
for f in sorted(os.listdir(dist_dir)):