Fork of Cyber security for passwords
pythonv1MITنُشر ١٩ أغسطس ٢٠٢٦· ♥ 0 · ⬇ 1 · 💬 0
كود ينشئ كلمات مرور و يحمي جهازك من attacks
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
SOVEREIGN-X PORTABLE SECURITY & SIGNAL ANALYZER v5.0
=====================================================
Production-grade single-file security tool for small devices.
SECURITY FEATURES:
- AES-256-GCM authenticated encryption
- PBKDF2-HMAC-SHA256 (600k iterations) or optional Argon2id
- hmac.compare_digest for constant-time comparison
- Auto-lock after inactivity
- Panic mode with self-destruct
- Encrypted export/import
- Isolated test suite via tempfile
USAGE:
python sovereignx.py # Run normally
python sovereignx.py --test # Run isolated tests
REQUIREMENTS:
pip install cryptography
# Optional for Argon2id:
pip install argon2-cffi
"""
import os
import sys
import json
import time
import math
import sqlite3
import hashlib
import secrets
import string
import logging
import getpass
import subprocess
import re
import threading
import hmac
import tempfile
from datetime import datetime
from pathlib import Path
from dataclasses import dataclass
from typing import List, Dict, Optional, Tuple
from contextlib import contextmanager
# ============================================================
# CRYPTOGRAPHY IMPORTS
# ============================================================
try:
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
from cryptography.hazmat.primitives import hashes
CRYPTO_AVAILABLE = True
except ImportError:
CRYPTO_AVAILABLE = False
print("[FATAL] cryptography library not installed!")
print("Run: pip install cryptography")
sys.exit(1)
# Optional Argon2id support via argon2-cffi
try:
from argon2.low_level import hash_secret_raw, Type
ARGON2_AVAILABLE = True
except ImportError:
ARGON2_AVAILABLE = False
# ============================================================
# CONFIGURATION
# ============================================================
class Config:
APP_NAME = "SovereignX-Security"
VERSION = "5.0.0"
BASE_DIR = Path.home() / ".sovereignx"
DB_PATH = BASE_DIR / "sovereignx.db"
LOG_PATH = BASE_DIR / "logs" / "sovereignx.log"
KEY_FILE = BASE_DIR / ".master_key"
# KDF Settings
# PBKDF2 is the default (NIST-approved, widely compatible)
# Argon2id is used if argon2-cffi is installed AND algorithm is set to "argon2id"
KDF_ALGORITHM = "pbkdf2" # "pbkdf2" or "argon2id"
PBKDF2_ITERATIONS = 600_000 # OWASP recommendation for PBKDF2-HMAC-SHA256
ARGON2_TIME_COST = 3 # Argon2id iterations
ARGON2_MEMORY = 65536 # KiB (64 MB)
ARGON2_PARALLELISM = 4 # Threads
# App settings
LOCK_TIMEOUT = 300 # seconds
SIGNAL_SCAN_INTERVAL = 5
MAX_SIGNAL_HISTORY = 1000
SCREEN_WIDTH = 40
SCREEN_HEIGHT = 20
MAX_FAILED_LOGINS = 5 # Self-destruct trigger
@classmethod
def ensure_dirs(cls):
cls.BASE_DIR.mkdir(parents=True, exist_ok=True)
(cls.BASE_DIR / "logs").mkdir(parents=True, exist_ok=True)
# ============================================================
# LOGGING
# ============================================================
Config.ensure_dirs()
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(message)s",
handlers=[
logging.FileHandler(Config.LOG_PATH),
logging.StreamHandler(sys.stdout)
]
)
logger = logging.getLogger(__name__)
# ============================================================
# DATABASE
# ============================================================
class Database:
"""
SQLite database with metadata table for KDF versioning and panic code storage.
Supports custom base_dir for isolated testing.
"""
def __init__(self, base_dir: Path = None):
if base_dir is None:
base_dir = Config.BASE_DIR
self.base_dir = Path(base_dir)
self.base_dir.mkdir(parents=True, exist_ok=True)
self.db_path = self.base_dir / "sovereignx.db"
self._init_db()
@contextmanager
def _connect(self):
conn = sqlite3.connect(str(self.db_path))
conn.row_factory = sqlite3.Row
conn.execute("PRAGMA foreign_keys = ON")
try:
yield conn
conn.commit()
except Exception as e:
conn.rollback()
logger.error(f"Database error: {e}")
raise
finally:
conn.close()
def _init_db(self):
with self._connect() as conn:
conn.execute("""
CREATE TABLE IF NOT EXISTS db_metadata (
key TEXT PRIMARY KEY,
value TEXT NOT NULL
)
""")
conn.execute("""
CREATE TABLE IF NOT EXISTS passwords (
id INTEGER PRIMARY KEY AUTOINCREMENT,
service TEXT NOT NULL,
username TEXT,
encrypted_password TEXT NOT NULL,
notes TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
conn.execute("""
CREATE TABLE IF NOT EXISTS signal_history (
id INTEGER PRIMARY KEY AUTOINCREMENT,
signal_type TEXT NOT NULL,
ssid TEXT,
bssid TEXT,
rssi INTEGER,
frequency REAL,
estimated_distance REAL,
distance_confidence TEXT,
timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
conn.execute("""
CREATE TABLE IF NOT EXISTS map_points (
id INTEGER PRIMARY KEY AUTOINCREMENT,
latitude REAL,
longitude REAL,
altitude REAL,
label TEXT,
signal_id INTEGER,
timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (signal_id) REFERENCES signal_history(id)
)
""")
conn.execute("""
CREATE TABLE IF NOT EXISTS auth_log (
id INTEGER PRIMARY KEY AUTOINCREMENT,
action TEXT NOT NULL,
status TEXT NOT NULL,
timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
logger.info("Database initialized")
def get_metadata(self, key: str) -> Optional[str]:
with self._connect() as conn:
row = conn.execute(
"SELECT value FROM db_metadata WHERE key = ?", (key,)
).fetchone()
return row["value"] if row else None
def set_metadata(self, key: str, value: str):
with self._connect() as conn:
conn.execute("""
INSERT INTO db_metadata (key, value) VALUES (?, ?)
ON CONFLICT(key) DO UPDATE SET value = excluded.value
""", (key, value))
def cleanup_old_signals(self):
with self._connect() as conn:
conn.execute("""
DELETE FROM signal_history WHERE id NOT IN
(SELECT id FROM signal_history ORDER BY timestamp DESC LIMIT ?)
""", (Config.MAX_SIGNAL_HISTORY,))
def wipe_all_data(self):
"""Securely wipe all tables - Panic/Self-destruct mode."""
with self._connect() as conn:
conn.execute("DELETE FROM passwords")
conn.execute("DELETE FROM signal_history")
conn.execute("DELETE FROM map_points")
conn.execute("DELETE FROM auth_log")
conn.execute("DELETE FROM db_metadata")
logger.critical("ALL DATABASE DATA WIPED")
def export_encrypted(self, encryption) -> dict:
"""Export all data as an encrypted JSON bundle."""
with self._connect() as conn:
passwords = [dict(r) for r in conn.execute("SELECT * FROM passwords").fetchall()]
signals = [dict(r) for r in conn.execute("SELECT * FROM signal_history").fetchall()]
points = [dict(r) for r in conn.execute("SELECT * FROM map_points").fetchall()]
meta = {r["key"]: r["value"] for r in conn.execute("SELECT * FROM db_metadata").fetchall()}
bundle = {
"version": Config.VERSION,
"exported_at": datetime.now().isoformat(),
"metadata": meta,
"passwords": passwords,
"signals": signals,
"points": points
}
return {"encrypted": encryption.encrypt(json.dumps(bundle, ensure_ascii=False))}
def import_encrypted(self, encrypted_bundle: dict, encryption):
"""Import encrypted JSON bundle. Ignores original IDs to avoid conflicts."""
bundle = json.loads(encryption.decrypt(encrypted_bundle["encrypted"]))
with self._connect() as conn:
for p in bundle.get("passwords", []):
conn.execute("""
INSERT INTO passwords (service, username, encrypted_password, notes, created_at, updated_at)
VALUES (?, ?, ?, ?, ?, ?)
""", (p["service"], p.get("username"), p["encrypted_password"],
p.get("notes"), p.get("created_at"), p.get("updated_at")))
for s in bundle.get("signals", []):
conn.execute("""
INSERT INTO signal_history (signal_type, ssid, bssid, rssi, frequency,
estimated_distance, distance_confidence, timestamp)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
""", (s["signal_type"], s.get("ssid"), s.get("bssid"), s.get("rssi"),
s.get("frequency"), s.get("estimated_distance"), s.get("distance_confidence"),
s.get("timestamp")))
for p in bundle.get("points", []):
conn.execute("""
INSERT INTO map_points (latitude, longitude, altitude, label, signal_id, timestamp)
VALUES (?, ?, ?, ?, ?, ?)
""", (p.get("latitude"), p.get("longitude"), p.get("altitude"),
p.get("label"), p.get("signal_id"), p.get("timestamp")))
logger.info(f"Imported {len(bundle.get('passwords', []))} passwords, "
f"{len(bundle.get('signals', []))} signals, "
f"{len(bundle.get('points', []))} points")
# ============================================================
# ENCRYPTION (AES-256-GCM)
# ============================================================
class Encryption:
"""
AES-256-GCM authenticated encryption.
Key derived via PBKDF2-HMAC-SHA256 (600k iterations) or Argon2id.
Salt and KDF metadata stored in database for migration support.
"""
def __init__(self, master_password: str, db: Database):
self.db = db
self.salt = self._load_or_create_salt()
self.key = self._derive_key(master_password, self.salt)
def _load_or_create_salt(self) -> bytes:
salt_b64 = self.db.get_metadata("salt")
if salt_b64:
import base64
return base64.b64decode(salt_b64)
else:
salt = secrets.token_bytes(32)
import base64
self.db.set_metadata("salt", base64.b64encode(salt).decode())
self.db.set_metadata("kdf", Config.KDF_ALGORITHM)
self.db.set_metadata("cipher", "aes-256-gcm")
self.db.set_metadata("version", Config.VERSION)
logger.info("New database salt and metadata generated")
return salt
def _derive_key(self, password: str, salt: bytes) -> bytes:
"""
Derive encryption key using configured KDF.
Default: PBKDF2-HMAC-SHA256 with 600k iterations (OWASP recommendation).
Optional: Argon2id via argon2-cffi library.
"""
if Config.KDF_ALGORITHM == "argon2id" and ARGON2_AVAILABLE:
logger.info("Using Argon2id KDF")
return hash_secret_raw(
password.encode("utf-8"),
salt,
time_cost=Config.ARGON2_TIME_COST,
memory_cost=Config.ARGON2_MEMORY,
parallelism=Config.ARGON2_PARALLELISM,
hash_len=32,
type=Type.ID
)
# Default: PBKDF2-HMAC-SHA256
if Config.KDF_ALGORITHM == "argon2id" and not ARGON2_AVAILABLE:
logger.warning("argon2-cffi not installed, falling back to PBKDF2")
logger.info(f"Using PBKDF2-HMAC-SHA256 with {Config.PBKDF2_ITERATIONS} iterations")
kdf = PBKDF2HMAC(
algorithm=hashes.SHA256(),
length=32,
salt=salt,
iterations=Config.PBKDF2_ITERATIONS,
)
return kdf.derive(password.encode("utf-8"))
def encrypt(self, data: str) -> str:
import base64
nonce = secrets.token_bytes(12)
aesgcm = AESGCM(self.key)
ciphertext = aesgcm.encrypt(nonce, data.encode("utf-8"), None)
return base64.b64encode(nonce + ciphertext).decode()
def decrypt(self, encrypted_data: str) -> str:
import base64
try:
data = base64.b64decode(encrypted_data.encode())
nonce, ciphertext = data[:12], data[12:]
aesgcm = AESGCM(self.key)
return aesgcm.decrypt(nonce, ciphertext, None).decode("utf-8")
except Exception as e:
logger.error(f"Decryption failed: {e}")
raise ValueError("Decryption failed - wrong password or corrupted data")
def clear_key(self):
"""Securely clear key from memory."""
self.key = None
self.salt = None
# ============================================================
# AUTHENTICATION (with Panic Code)
# ============================================================
class Authentication:
"""
Handles master password verification with:
- PBKDF2-HMAC-SHA256 password hashing
- hmac.compare_digest for timing-attack resistance
- Configurable panic code (stored as hash in DB metadata)
- Auto-lock after inactivity
- Self-destruct after max failed logins
"""
def __init__(self, db: Database):
self.db = db
self._master_hash = None
self._panic_hash = None
self._last_activity = time.time()
self._locked = False
self._failed_attempts = 0
self._load_hashes()
def _load_hashes(self):
key_file = self.db.base_dir / ".master_key"
if key_file.exists():
with open(key_file, "r", encoding="utf-8") as f:
self._master_hash = f.read().strip()
# Load panic hash from DB metadata
self._panic_hash = self.db.get_metadata("panic_hash")
def _hash_password(self, password: str) -> str:
salt = secrets.token_hex(16)
pwdhash = hashlib.pbkdf2_hmac("sha256", password.encode(), salt.encode(), 100000)
return salt + pwdhash.hex()
def _verify_password(self, password: str, stored: str) -> bool:
salt = stored[:32]
stored_hash = stored[32:]
pwdhash = hashlib.pbkdf2_hmac("sha256", password.encode(), salt.encode(), 100000)
return hmac.compare_digest(pwdhash.hex(), stored_hash)
def is_first_run(self) -> bool:
return self._master_hash is None
def setup_master_password(self, password: str, panic_code: str):
if len(password) < 8:
raise ValueError("Password must be at least 8 characters")
if len(panic_code) < 4:
raise ValueError("Panic code must be at least 4 characters")
self._master_hash = self._hash_password(password)
key_file = self.db.base_dir / ".master_key"
with open(key_file, "w", encoding="utf-8") as f:
f.write(self._master_hash)
if os.name != "nt":
os.chmod(key_file, 0o600)
# Store panic code hash in DB (not plaintext)
self._panic_hash = self._hash_password(panic_code)
self.db.set_metadata("panic_hash", self._panic_hash)
self._log_auth("SETUP", "SUCCESS")
logger.info("Master password and panic code set")
def login(self, password: str) -> Tuple[bool, bool]:
"""
Returns: (success, is_panic)
is_panic=True triggers self-destruct.
"""
# Check panic code first
if self._panic_hash and self._verify_password(password, self._panic_hash):
logger.critical("PANIC CODE ENTERED")
return False, True
if self._master_hash is None:
return False, False
if self._verify_password(password, self._master_hash):
self._last_activity = time.time()
self._locked = False
self._failed_attempts = 0
self._log_auth("LOGIN", "SUCCESS")
return True, False
self._failed_attempts += 1
self._log_auth("LOGIN", "FAILED")
logger.warning(f"Failed attempt {self._failed_attempts}/{Config.MAX_FAILED_LOGINS}")
if self._failed_attempts >= Config.MAX_FAILED_LOGINS:
logger.critical("MAX FAILED LOGINS - SELF-DESTRUCT")
return False, True
return False, False
def check_auto_lock(self):
if time.time() - self._last_activity > Config.LOCK_TIMEOUT:
if not self._locked:
self._locked = True
logger.info("Auto-locked")
return self._locked
def update_activity(self):
self._last_activity = time.time()
self._locked = False
def _log_auth(self, action: str, status: str):
with self.db._connect() as conn:
conn.execute(
"INSERT INTO auth_log (action, status) VALUES (?, ?)",
(action, status)
)
# ============================================================
# SECURITY MANAGER
# ============================================================
class SecurityManager:
"""Central security coordinator managing auth, encryption, and session lifecycle."""
def __init__(self, db: Database):
self.db = db
self.auth = Authentication(db)
self.encryption = None
self._session_active = False
def is_first_run(self) -> bool:
return self.auth.is_first_run()
def setup(self, password: str, panic_code: str):
self.auth.setup_master_password(password, panic_code)
self._start_session(password)
def login(self, password: str) -> Tuple[bool, bool]:
success, is_panic = self.auth.login(password)
if is_panic:
self._trigger_panic()
return False, True
if success:
self._start_session(password)
return True, False
return False, False
def _trigger_panic(self):
logger.critical("PANIC MODE - WIPING ALL DATA")
if self.encryption:
self.encryption.clear_key()
self.encryption = None
self.db.wipe_all_data()
key_file = self.db.base_dir / ".master_key"
if key_file.exists():
key_file.unlink()
self._session_active = False
def _start_session(self, password: str):
self.encryption = Encryption(password, self.db)
self._session_active = True
logger.info("Session started")
def lock(self):
if self.encryption:
self.encryption.clear_key()
self.encryption = None
self._session_active = False
self.auth._locked = True
logger.info("Session locked")
def unlock(self, password: str) -> Tuple[bool, bool]:
success, is_panic = self.auth.login(password)
if is_panic:
self._trigger_panic()
return False, True
if success:
self._start_session(password)
return True, False
return False, False
def check_auto_lock(self):
if self.auth.check_auto_lock() and self._session_active:
self.lock()
return self.auth._locked
def update_activity(self):
self.auth.update_activity()
@property
def is_locked(self) -> bool:
return self.auth._locked or not self._session_active
@property
def is_active(self) -> bool:
return self._session_active and not self.auth._locked
# ============================================================
# PASSWORD MANAGER
# ============================================================
class PasswordManager:
def __init__(self, db: Database, encryption: Encryption):
self.db = db
self.encryption = encryption
def add_password(self, service: str, username: str, password: str, notes: str = ""):
encrypted = self.encryption.encrypt(password)
with self.db._connect() as conn:
conn.execute("""
INSERT INTO passwords (service, username, encrypted_password, notes)
VALUES (?, ?, ?, ?)
""", (service, username, encrypted, notes))
logger.info(f"Password added: {service}")
def get_passwords(self) -> List[Dict]:
with self.db._connect() as conn:
rows = conn.execute("SELECT * FROM passwords ORDER BY created_at DESC").fetchall()
return [dict(row) for row in rows]
def get_password(self, entry_id: int) -> Optional[str]:
with self.db._connect() as conn:
row = conn.execute(
"SELECT encrypted_password FROM passwords WHERE id = ?", (entry_id,)
).fetchone()
if row:
return self.encryption.decrypt(row["encrypted_password"])
return None
def delete_password(self, entry_id: int):
with self.db._connect() as conn:
conn.execute("DELETE FROM passwords WHERE id = ?", (entry_id,))
logger.info(f"Deleted entry {entry_id}")
def update_password(self, entry_id: int, service: str = None, username: str = None,
password: str = None, notes: str = None):
updates, params = [], []
if service is not None:
updates.append("service = ?")
params.append(service)
if username is not None:
updates.append("username = ?")
params.append(username)
if password is not None:
updates.append("encrypted_password = ?")
params.append(self.encryption.encrypt(password))
if notes is not None:
updates.append("notes = ?")
params.append(notes)
if updates:
params.append(entry_id)
with self.db._connect() as conn:
conn.execute(
f"UPDATE passwords SET {', '.join(updates)}, updated_at = CURRENT_TIMESTAMP WHERE id = ?",
params
)
logger.info(f"Updated entry {entry_id}")
# ============================================================
# PASSWORD GENERATOR
# ============================================================
class PasswordGenerator:
@staticmethod
def generate(length: int = 16, uppercase: bool = True, lowercase: bool = True,
digits: bool = True, symbols: bool = True) -> Tuple[str, str]:
if length < 4:
raise ValueError("Length must be at least 4")
categories = []
if lowercase: categories.append(string.ascii_lowercase)
if uppercase: categories.append(string.ascii_uppercase)
if digits: categories.append(string.digits)
if symbols: categories.append("!@#$%^&*()_+-=[]{}|;:,.<>?")
if not categories:
raise ValueError("Select at least one character type")
if length < len(categories):
raise ValueError(f"Length must be at least {len(categories)}")
password_chars = [secrets.choice(cat) for cat in categories]
all_chars = "".join(categories)
for _ in range(length - len(categories)):
password_chars.append(secrets.choice(all_chars))
shuffled = list(password_chars)
for i in range(len(shuffled) - 1, 0, -1):
j = secrets.randbelow(i + 1)
shuffled[i], shuffled[j] = shuffled[j], shuffled[i]
password = "".join(shuffled)
strength = PasswordGenerator._check_strength(password)
return password, strength
@staticmethod
def _check_strength(password: str) -> str:
score = 0
if len(password) >= 12: score += 1
if len(password) >= 16: score += 1
if any(c.isupper() for c in password): score += 1
if any(c.islower() for c in password): score += 1
if any(c.isdigit() for c in password): score += 1
if any(c in string.punctuation for c in password): score += 1
if score <= 2: return "Weak"
elif score <= 4: return "Medium"
elif score <= 5: return "Strong"
else: return "Very Strong"
# ============================================================
# SIGNAL SCANNER
# ============================================================
@dataclass
class SignalReading:
signal_type: str
ssid: Optional[str]
bssid: str
rssi: int
frequency: Optional[float]
estimated_distance: Optional[float]
distance_confidence: str
timestamp: str
class SignalScanner:
"""
Scan WiFi and Bluetooth signals.
Distance is estimated via Free Space Path Loss - APPROXIMATE ONLY.
"""
def __init__(self, db: Database):
self.db = db
self.scanning = False
self._scan_thread = None
self._last_results = []
def _estimate_distance(self, rssi: int, freq_mhz: float = 2400) -> Tuple[float, str]:
"""Estimate distance using FSPL. Returns (meters, confidence)."""
try:
distance = 10 ** ((27.55 - 20 * math.log10(freq_mhz) + abs(rssi)) / 20)
distance = round(min(distance, 100), 2)
if rssi > -50: confidence = "HIGH"
elif rssi > -70: confidence = "MEDIUM"
else: confidence = "LOW"
return distance, confidence
except:
return -1.0, "UNKNOWN"
def scan_wifi(self) -> List[SignalReading]:
results = []
try:
cmd = ["nmcli", "-t", "-f", "SSID,BSSID,SIGNAL,FREQ",
"--separator", "|", "dev", "wifi"]
output = subprocess.run(cmd, capture_output=True, text=True, timeout=10)
if output.returncode == 0 and output.stdout:
for line in output.stdout.strip().split(chr(10)):
parts = line.split("|")
if len(parts) >= 4:
ssid = parts[0] if parts[0] else "Hidden"
bssid = parts[1]
signal = int(parts[2]) if parts[2].isdigit() else -100
freq = float(parts[3]) if parts[3] else 2400
rssi_estimated = (signal / 2) - 100
distance, confidence = self._estimate_distance(int(rssi_estimated), freq)
results.append(SignalReading(
signal_type="WiFi", ssid=ssid[:20], bssid=bssid,
rssi=int(rssi_estimated), frequency=freq,
estimated_distance=distance, distance_confidence=confidence,
timestamp=datetime.now().isoformat()
))
except FileNotFoundError:
logger.debug("nmcli not found")
except Exception as e:
logger.error(f"nmcli error: {e}")
if not results:
try:
cmd = ["sudo", "iwlist", "scan"]
output = subprocess.run(cmd, capture_output=True, text=True, timeout=15)
if output.returncode == 0:
results = self._parse_iwlist(output.stdout)
except FileNotFoundError:
logger.warning("iwlist not found")
except Exception as e:
logger.error(f"iwlist error: {e}")
return results
def _parse_iwlist(self, output: str) -> List[SignalReading]:
results = []
cells = output.split("Cell ")
for cell in cells[1:]:
try:
essid_match = re.search(r'ESSID:"([^"]*)"', cell)
address_match = re.search(r'Address: ([0-9A-F:]{17})', cell)
signal_match = re.search(r'Signal level=(-?\d+)', cell)
freq_match = re.search(r'Frequency:([\d.]+)', cell)
if address_match:
ssid = essid_match.group(1) if essid_match else "Hidden"
bssid = address_match.group(1)
rssi = int(signal_match.group(1)) if signal_match else -100
freq = float(freq_match.group(1)) * 1000 if freq_match else 2400
distance, confidence = self._estimate_distance(rssi, freq)
results.append(SignalReading(
signal_type="WiFi", ssid=ssid[:20], bssid=bssid,
rssi=rssi, frequency=freq,
estimated_distance=distance, distance_confidence=confidence,
timestamp=datetime.now().isoformat()
))
except Exception:
continue
return results
def scan_bluetooth(self) -> List[SignalReading]:
results = []
btctl_results = self._scan_bluetoothctl()
if btctl_results:
return btctl_results
try:
cmd = ["hcitool", "scan", "--flush"]
output = subprocess.run(cmd, capture_output=True, text=True, timeout=10)
if output.returncode == 0:
for line in output.stdout.strip().split(chr(10))[1:]:
parts = line.strip().split(chr(9))
if len(parts) >= 2:
mac = parts[0]
name = parts[1] if len(parts) > 1 else "Unknown"
rssi = -100
try:
rssi_cmd = ["hcitool", "rssi", mac]
rssi_out = subprocess.run(rssi_cmd, capture_output=True, text=True, timeout=5)
if rssi_out.returncode == 0:
rssi_match = re.search(r'-?\d+', rssi_out.stdout)
if rssi_match:
rssi = int(rssi_match.group())
except:
pass
distance, confidence = self._estimate_distance(rssi, 2400)
results.append(SignalReading(
signal_type="Bluetooth", ssid=name[:20], bssid=mac,
rssi=rssi, frequency=2400,
estimated_distance=distance, distance_confidence=confidence,
timestamp=datetime.now().isoformat()
))
except FileNotFoundError:
logger.warning("hcitool not found")
except Exception as e:
logger.error(f"Bluetooth scan error: {e}")
return results
def _scan_bluetoothctl(self) -> List[SignalReading]:
results = []
try:
subprocess.run(["bluetoothctl", "scan", "on"],
capture_output=True, text=True, timeout=2)
time.sleep(5)
output = subprocess.run(["bluetoothctl", "devices"],
capture_output=True, text=True, timeout=5)
if output.returncode == 0:
for line in output.stdout.strip().split(chr(10)):
match = re.match(r'Device\s+([0-9A-F:]{17})\s+(.+)', line)
if match:
mac = match.group(1)
name = match.group(2)
rssi = -100
try:
info = subprocess.run(["bluetoothctl", "info", mac],
capture_output=True, text=True, timeout=3)
rssi_match = re.search(r'RSSI:\s*(-?\d+)', info.stdout)
if rssi_match:
rssi = int(rssi_match.group(1))
except:
pass
distance, confidence = self._estimate_distance(rssi, 2400)
results.append(SignalReading(
signal_type="Bluetooth", ssid=name[:20], bssid=mac,
rssi=rssi, frequency=2400,
estimated_distance=distance, distance_confidence=confidence,
timestamp=datetime.now().isoformat()
))
subprocess.run(["bluetoothctl", "scan", "off"],
capture_output=True, text=True, timeout=2)
except FileNotFoundError:
logger.debug("bluetoothctl not found")
except Exception as e:
logger.error(f"bluetoothctl error: {e}")
return results
def scan_all(self) -> List[SignalReading]:
wifi = self.scan_wifi()
bt = self.scan_bluetooth()
all_signals = wifi + bt
all_signals.sort(key=lambda x: x.rssi, reverse=True)
self._last_results = all_signals
self._save_to_db(all_signals)
self.db.cleanup_old_signals()
return all_signals
def _save_to_db(self, signals: List[SignalReading]):
with self.db._connect() as conn:
for sig in signals:
conn.execute("""
INSERT INTO signal_history
(signal_type, ssid, bssid, rssi, frequency, estimated_distance, distance_confidence)
VALUES (?, ?, ?, ?, ?, ?, ?)
""", (sig.signal_type, sig.ssid, sig.bssid, sig.rssi,
sig.frequency, sig.estimated_distance, sig.distance_confidence))
def get_history(self, limit: int = 50) -> List[Dict]:
with self.db._connect() as conn:
rows = conn.execute(
"SELECT * FROM signal_history ORDER BY timestamp DESC LIMIT ?", (limit,)
).fetchall()
return [dict(row) for row in rows]
def start_continuous_scan(self, callback=None):
self.scanning = True
def scan_loop():
while self.scanning:
results = self.scan_all()
if callback:
callback(results)
time.sleep(Config.SIGNAL_SCAN_INTERVAL)
self._scan_thread = threading.Thread(target=scan_loop, daemon=True)
self._scan_thread.start()
def stop_continuous_scan(self):
self.scanning = False
# ============================================================
# MAP MANAGER
# ============================================================
class MapManager:
def __init__(self, db: Database):
self.db = db
def get_gps_location(self) -> Optional[Tuple[float, float, float]]:
loc = self._get_gpsd_location()
if loc: return loc
loc = self._get_geoclue_location()
if loc: return loc
loc = self._get_termux_location()
if loc: return loc
return None
def _get_gpsd_location(self) -> Optional[Tuple[float, float, float]]:
try:
import socket
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(3)
sock.connect(("localhost", 2947))
sock.sendall(b'?WATCH={\"enable\":true}\n')
time.sleep(1)
sock.sendall(b'?POLL;\n')
data = b''
start_time = time.time()
while time.time() - start_time < 3:
chunk = sock.recv(4096)
if not chunk: break
data += chunk
if b'"lat"' in data and b'"lon"' in data: break
sock.close()
data_str = data.decode()
lat_match = re.search(r'"lat":([-?\d.]+)', data_str)
lon_match = re.search(r'"lon":([-?\d.]+)', data_str)
alt_match = re.search(r'"alt":([-?\d.]+)', data_str)
if lat_match and lon_match:
return (float(lat_match.group(1)), float(lon_match.group(1)),
float(alt_match.group(1)) if alt_match else 0.0)
except Exception as e:
logger.debug(f"gpsd error: {e}")
return None
def _get_geoclue_location(self) -> Optional[Tuple[float, float, float]]:
try:
client_cmd = ["busctl", "--user", "call", "org.freedesktop.GeoClue2",
"/org/freedesktop/GeoClue2/Manager", "org.freedesktop.GeoClue2.Manager", "GetClient"]
client_out = subprocess.run(client_cmd, capture_output=True, text=True, timeout=5)
if client_out.returncode == 0:
path_match = re.search(r'"(/org/freedesktop/GeoClue2/Client/\d+)"', client_out.stdout)
if path_match:
client_path = path_match.group(1)
subprocess.run(["busctl", "--user", "set-property", "org.freedesktop.GeoClue2",
client_path, "org.freedesktop.GeoClue2.Client", "DesktopId",
"s", "sovereignx"], capture_output=True, timeout=3)
subprocess.run(["busctl", "--user", "call", "org.freedesktop.GeoClue2",
client_path, "org.freedesktop.GeoClue2.Client", "Start"],
capture_output=True, timeout=3)
time.sleep(2)
loc_cmd = ["busctl", "--user", "get-property", "org.freedesktop.GeoClue2",
client_path, "org.freedesktop.GeoClue2.Client", "Location"]
loc_out = subprocess.run(loc_cmd, capture_output=True, text=True, timeout=3)
if loc_out.returncode == 0:
path_match = re.search(r'"(/org/freedesktop/GeoClue2/Location/\d+)"', loc_out.stdout)
if path_match:
loc_path = path_match.group(1)
props = ["Latitude", "Longitude", "Altitude"]
vals = []
for prop in props:
pcmd = ["busctl", "--user", "get-property", "org.freedesktop.GeoClue2",
loc_path, "org.freedesktop.GeoClue2.Location", prop]
pout = subprocess.run(pcmd, capture_output=True, text=True, timeout=3)
val_match = re.search(r'd\s+(-?[\d.]+)', pout.stdout)
vals.append(float(val_match.group(1)) if val_match else 0.0)
return tuple(vals)
except Exception as e:
logger.debug(f"geoclue error: {e}")
return None
def _get_termux_location(self) -> Optional[Tuple[float, float, float]]:
try:
output = subprocess.run(["termux-location"], capture_output=True, text=True, timeout=10)
if output.returncode == 0:
data = json.loads(output.stdout)
return (data.get("latitude", 0.0), data.get("longitude", 0.0), data.get("altitude", 0.0))
except (FileNotFoundError, json.JSONDecodeError):
pass
except Exception as e:
logger.debug(f"termux-location error: {e}")
return None
def add_point(self, lat: float, lon: float, altitude: float = 0.0, label: str = "", signal_id: int = None):
with self.db._connect() as conn:
conn.execute("""
INSERT INTO map_points (latitude, longitude, altitude, label, signal_id)
VALUES (?, ?, ?, ?, ?)
""", (lat, lon, altitude, label, signal_id))
logger.info(f"Map point added: {lat}, {lon}")
def get_points(self, limit: int = 100) -> List[Dict]:
with self.db._connect() as conn:
rows = conn.execute("SELECT * FROM map_points ORDER BY timestamp DESC LIMIT ?", (limit,)).fetchall()
return [dict(row) for row in rows]
def generate_ascii_map(self, points: List[Dict], width: int = 40, height: int = 15) -> str:
if not points:
return "No points on map"
lats = [p["latitude"] for p in points]
lons = [p["longitude"] for p in points]
min_lat, max_lat = min(lats), max(lats)
min_lon, max_lon = min(lons), max(lons)
lat_pad = (max_lat - min_lat) * 0.1 or 0.001
lon_pad = (max_lon - min_lon) * 0.1 or 0.001
min_lat -= lat_pad; max_lat += lat_pad
min_lon -= lon_pad; max_lon += lon_pad
grid = [["." for _ in range(width)] for _ in range(height)]
for p in points:
x = int((p["longitude"] - min_lon) / (max_lon - min_lon) * (width - 1))
y = int((max_lat - p["latitude"]) / (max_lat - min_lat) * (height - 1))
x = max(0, min(x, width - 1))
y = max(0, min(y, height - 1))
grid[y][x] = "X"
lines = [
"+" + "-" * width + "+",
f"|{"Map Points":^{width}}|",
"+" + "-" * width + "+"
]
for row in grid:
lines.append("|" + "".join(row) + "|")
lines.append("+" + "-" * width + "+")
lines.append(f"Range: [{min_lat:.4f}, {min_lon:.4f}] -> [{max_lat:.4f}, {max_lon:.4f}]")
return chr(10).join(lines)
# ============================================================
# TERMINAL UI
# ============================================================
class TerminalUI:
def __init__(self, security: SecurityManager, db: Database):
self.security = security
self.db = db
self.password_manager = None
self.scanner = SignalScanner(db)
self.map_manager = MapManager(db)
self.running = True
def clear(self):
os.system("clear" if os.name != "nt" else "cls")
def print_header(self, title: str):
w = Config.SCREEN_WIDTH
self.clear()
print("=" * w)
print(f"{title:^{w}}")
print("=" * w)
def print_box(self, text: str):
w = Config.SCREEN_WIDTH
print("+" + "-" * (w - 2) + "+")
for line in text.split(chr(10)):
print(f"| {line:<{w-4}} |")
print("+" + "-" * (w - 2) + "+")
def input_prompt(self, prompt: str) -> str:
self.security.update_activity()
value = input(f">>> {prompt}: ").strip()
self.security.update_activity()
return value
def secure_input(self, prompt: str) -> str:
self.security.update_activity()
value = getpass.getpass(f">>> {prompt}: ")
self.security.update_activity()
return value
def show_menu(self, title: str, options: Dict[str, str]) -> str:
self.print_header(title)
for key, value in options.items():
print(f" [{key}] {value}")
print("-" * Config.SCREEN_WIDTH)
return self.input_prompt("Select")
def run(self):
if self.security.is_first_run():
self.setup_wizard()
if not self.login_screen():
return
while self.running:
if self.security.check_auto_lock():
self.locked_screen()
continue
choice = self.show_menu("SovereignX Security Hub", {
"1": "Password Vault",
"2": "Password Generator",
"3": "Signal Scanner",
"4": "Signal Analyzer",
"5": "GPS / Map",
"6": "Activity Log",
"7": "Export / Import",
"8": "Lock Device",
"0": "Exit"
})
if choice == "1": self.password_manager_menu()
elif choice == "2": self.password_generator_menu()
elif choice == "3": self.signal_scanner_menu()
elif choice == "4": self.signal_analyzer_menu()
elif choice == "5": self.map_menu()
elif choice == "6": self.activity_log_menu()
elif choice == "7": self.export_import_menu()
elif choice == "8": self.security.lock()
elif choice == "0": self.running = False
def setup_wizard(self):
self.print_header("Initial Setup")
print("Welcome to SovereignX Security Hub")
print("Create master password and panic code")
print("-" * Config.SCREEN_WIDTH)
while True:
pwd = self.secure_input("Master password (min 8 chars)")
confirm = self.secure_input("Confirm password")
if pwd != confirm or len(pwd) < 8:
print("Error: Passwords do not match or too short")
continue
panic = self.secure_input("Panic code (min 4 chars, used to wipe data)")
if len(panic) < 4:
print("Error: Panic code too short")
continue
try:
self.security.setup(pwd, panic)
self.password_manager = PasswordManager(self.db, self.security.encryption)
print("Setup complete!")
time.sleep(1)
break
except Exception as e:
print(f"Error: {e}")
def login_screen(self) -> bool:
attempts = 0
while attempts < 3:
self.print_header("Login")
pwd = self.secure_input("Master password")
success, panic = self.security.login(pwd)
if panic:
self.print_header("PANIC MODE ACTIVATED")
print("All data has been wiped!")
print("Restart required...")
time.sleep(3)
sys.exit(0)
if success:
self.password_manager = PasswordManager(self.db, self.security.encryption)
print("Login successful")
time.sleep(1)
return True
else:
attempts += 1
print(f"Wrong password (attempt {attempts}/3)")
time.sleep(1)
print("Max attempts exceeded. Exiting...")
time.sleep(2)
return False
def locked_screen(self):
self.print_header("Device Locked")
print("Auto-locked due to inactivity")
print("-" * Config.SCREEN_WIDTH)
pwd = self.secure_input("Enter password to unlock")
success, panic = self.security.unlock(pwd)
if panic:
self.print_header("PANIC MODE ACTIVATED")
print("All data has been wiped!")
time.sleep(3)
sys.exit(0)
if success:
self.password_manager = PasswordManager(self.db, self.security.encryption)
print("Unlocked")
time.sleep(1)
else:
print("Wrong password")
time.sleep(1)
def password_manager_menu(self):
while True:
choice = self.show_menu("Password Vault", {
"1": "View saved passwords",
"2": "Add new password",
"3": "Delete password",
"4": "Show password (decrypt)",
"0": "Back"
})
if choice == "1":
passwords = self.password_manager.get_passwords()
self.print_header("Saved Passwords")
if not passwords:
print("No passwords saved")
else:
for p in passwords:
print(f"ID:{p['id']} | {p['service']} | {p['username']} | {p['created_at']}")
input("Press Enter to continue...")
elif choice == "2":
service = self.input_prompt("Service name")
username = self.input_prompt("Username")
password = self.secure_input("Password")
notes = self.input_prompt("Notes (optional)")
self.password_manager.add_password(service, username, password, notes)
print("Saved successfully")
time.sleep(1)
elif choice == "3":
pid = self.input_prompt("Enter ID to delete")
if pid.isdigit():
self.password_manager.delete_password(int(pid))
print("Deleted")
time.sleep(1)
elif choice == "4":
pid = self.input_prompt("Enter ID to show password")
if pid.isdigit():
pwd = self.password_manager.get_password(int(pid))
if pwd:
self.print_box(f"Password: {pwd}")
else:
print("Not found")
input("Press Enter to continue...")
elif choice == "0":
break
def password_generator_menu(self):
self.print_header("Password Generator")
try:
length = int(self.input_prompt("Password length (default 16)") or "16")
upper = self.input_prompt("Include uppercase? (y/n)").lower() != "n"
lower = self.input_prompt("Include lowercase? (y/n)").lower() != "n"
digits = self.input_prompt("Include digits? (y/n)").lower() != "n"
symbols = self.input_prompt("Include symbols? (y/n)").lower() != "n"
password, strength = PasswordGenerator.generate(length, upper, lower, digits, symbols)
self.print_box(f"Password: {password}\nStrength: {strength}")
save = self.input_prompt("Save to vault? (y/n)").lower()
if save == "y":
service = self.input_prompt("Service name")
username = self.input_prompt("Username")
self.password_manager.add_password(service, username, password)
print("Saved")
except Exception as e:
print(f"Error: {e}")
input("Press Enter to continue...")
def signal_scanner_menu(self):
while True:
choice = self.show_menu("Signal Scanner", {
"1": "Scan WiFi",
"2": "Scan Bluetooth",
"3": "Full Scan",
"4": "Continuous Scan",
"5": "Stop Continuous",
"0": "Back"
})
if choice == "1":
self.print_header("Scanning WiFi...")
results = self.scanner.scan_wifi()
self._display_signals(results)
elif choice == "2":
self.print_header("Scanning Bluetooth...")
results = self.scanner.scan_bluetooth()
self._display_signals(results)
elif choice == "3":
self.print_header("Full Scan...")
results = self.scanner.scan_all()
self._display_signals(results)
elif choice == "4":
print("Starting continuous scan... (Press Enter to stop)")
self.scanner.start_continuous_scan(self._display_signals_callback)
input()
self.scanner.stop_continuous_scan()
elif choice == "5":
self.scanner.stop_continuous_scan()
print("Scan stopped")
time.sleep(1)
elif choice == "0":
break
def _display_signals(self, results: List[SignalReading]):
self.print_header(f"Scan Results - {len(results)} signals")
if not results:
print("No signals found")
else:
print(f"{'Type':<8} {'Name':<15} {'RSSI':<6} {'Distance':<10}")
print("-" * Config.SCREEN_WIDTH)
for r in results[:20]:
dist = f"{r.estimated_distance}m" if r.estimated_distance and r.estimated_distance > 0 else "N/A"
name = (r.ssid or "Hidden")[:14]
print(f"{r.signal_type:<8} {name:<15} {r.rssi:<6} {dist:<10}")
print("\n[!] Distance is approximate - affected by walls and interference")
input("Press Enter to continue...")
def _display_signals_callback(self, results: List[SignalReading]):
self.clear()
print(f"[Continuous] {len(results)} signals detected")
for r in results[:10]:
dist = f"{r.estimated_distance}m" if r.estimated_distance and r.estimated_distance > 0 else "N/A"
conf = r.distance_confidence
print(f"{r.signal_type}: {r.ssid or 'Hidden'} | RSSI:{r.rssi} | {dist} ({conf})")
print("\n(Press Enter in menu to stop)")
def signal_analyzer_menu(self):
self.print_header("Signal Analyzer")
history = self.scanner.get_history(50)
if not history:
print("No data available")
else:
wifi_count = sum(1 for h in history if h["signal_type"] == "WiFi")
bt_count = sum(1 for h in history if h["signal_type"] == "Bluetooth")
avg_rssi = sum(h["rssi"] for h in history) / len(history)
self.print_box(f"Total readings: {len(history)}\nWiFi: {wifi_count} | BT: {bt_count}\nAvg RSSI: {avg_rssi:.1f} dBm")
print("\nLast 10 readings:")
for h in history[:10]:
dist = f"{h['estimated_distance']}m" if h["estimated_distance"] else "N/A"
conf = h.get("distance_confidence", "N/A")
print(f"{h['timestamp'][:16]} | {h['signal_type']} | {h['ssid'] or 'Hidden'} | {dist} ({conf})")
input("Press Enter to continue...")
def map_menu(self):
while True:
choice = self.show_menu("GPS / Map", {
"1": "Read current GPS location",
"2": "View saved points",
"3": "Add point manually",
"4": "Show ASCII map",
"0": "Back"
})
if choice == "1":
self.print_header("Reading GPS...")
loc = self.map_manager.get_gps_location()
if loc:
lat, lon, alt = loc
self.print_box(f"Current location:\nLat: {lat:.6f}\nLon: {lon:.6f}\nAlt: {alt:.1f}m")
save = self.input_prompt("Save point? (y/n)").lower()
if save == "y":
label = self.input_prompt("Point name")
self.map_manager.add_point(lat, lon, alt, label)
print("Saved")
else:
print("Cannot get GPS location")
print("Ensure gpsd or location service is running")
time.sleep(2)
elif choice == "2":
points = self.map_manager.get_points()
self.print_header("Saved Points")
if not points:
print("No points")
else:
for p in points:
print(f"ID:{p['id']} | {p['label'] or 'Unnamed'} | {p['latitude']:.4f}, {p['longitude']:.4f}")
input("Press Enter to continue...")
elif choice == "3":
try:
lat = float(self.input_prompt("Latitude"))
lon = float(self.input_prompt("Longitude"))
alt = float(self.input_prompt("Altitude (default 0)") or "0")
label = self.input_prompt("Point name")
self.map_manager.add_point(lat, lon, alt, label)
print("Added")
except ValueError:
print("Invalid coordinates")
time.sleep(1)
elif choice == "4":
points = self.map_manager.get_points()
self.print_header("ASCII Map")
print(self.map_manager.generate_ascii_map(points))
input("Press Enter to continue...")
elif choice == "0":
break
def export_import_menu(self):
while True:
choice = self.show_menu("Export / Import", {
"1": "Export encrypted backup",
"2": "Import encrypted backup",
"0": "Back"
})
if choice == "1":
self.print_header("Export Backup")
try:
bundle = self.db.export_encrypted(self.security.encryption)
filename = self.input_prompt("Filename (no extension)") or "sovereignx_backup"
filepath = self.db.base_dir / f"{filename}.json"
with open(filepath, "w", encoding="utf-8") as f:
json.dump(bundle, f, ensure_ascii=False)
print(f"Exported to: {filepath}")
except Exception as e:
print(f"Error: {e}")
time.sleep(1)
elif choice == "2":
self.print_header("Import Backup")
try:
filename = self.input_prompt("Filename (no extension)") or "sovereignx_backup"
filepath = self.db.base_dir / f"{filename}.json"
if not filepath.exists():
print("File not found")
time.sleep(1)
continue
confirm = self.input_prompt("This will REPLACE current data! Continue? (yes/no)").lower()
if confirm != "yes":
print("Cancelled")
time.sleep(1)
continue
with open(filepath, "r", encoding="utf-8") as f:
bundle = json.load(f)
self.db.wipe_all_data()
self.db.import_encrypted(bundle, self.security.encryption)
print("Import successful")
except Exception as e:
print(f"Error: {e}")
time.sleep(1)
elif choice == "0":
break
def activity_log_menu(self):
self.print_header("Activity Log")
with self.db._connect() as conn:
logs = conn.execute("SELECT * FROM auth_log ORDER BY timestamp DESC LIMIT 20").fetchall()
if not logs:
print("No logs")
else:
for log in logs:
status_icon = "OK" if log["status"] == "SUCCESS" else "FAIL"
print(f"{status_icon} {log['timestamp'][:16]} | {log['action']}")
input("Press Enter to continue...")
# ============================================================
# TESTS (Fully Isolated via tempfile)
# ============================================================
def run_tests():
print("\n" + "=" * 50)
print("RUNNING ISOLATED TESTS")
print("=" * 50 + "\n")
passed = 0
failed = 0
with tempfile.TemporaryDirectory(prefix="sovereignx_test_") as tmpdir:
test_base = Path(tmpdir)
print(f"Test directory: {test_base}\n")
# Test 1: Encryption roundtrip
try:
db = Database(base_dir=test_base)
enc = Encryption("TestPassword123!", db)
original = "Hello SovereignX"
encrypted = enc.encrypt(original)
decrypted = enc.decrypt(encrypted)
assert decrypted == original
assert encrypted != original
print("[PASS] test_encryption_roundtrip")
passed += 1
except Exception as e:
print(f"[FAIL] test_encryption_roundtrip: {e}")
failed += 1
# Test 2: Wrong password fails
try:
db = Database(base_dir=test_base)
enc1 = Encryption("CorrectPassword123!", db)
encrypted = enc1.encrypt("Secret Data")
enc2 = Encryption("WrongPassword456!", db)
try:
enc2.decrypt(encrypted)
print("[FAIL] test_wrong_password_fails: Should have raised error")
failed += 1
except ValueError:
print("[PASS] test_wrong_password_fails")
passed += 1
except Exception as e:
print(f"[FAIL] test_wrong_password_fails: {e}")
failed += 1
# Test 3: Password generator guarantees categories
try:
pwd, strength = PasswordGenerator.generate(16, uppercase=True, lowercase=True,
digits=True, symbols=True)
assert len(pwd) == 16
assert any(c.isupper() for c in pwd)
assert any(c.islower() for c in pwd)
assert any(c.isdigit() for c in pwd)
assert any(c in string.punctuation for c in pwd)
print("[PASS] test_password_generator_categories")
passed += 1
except Exception as e:
print(f"[FAIL] test_password_generator_categories: {e}")
failed += 1
# Test 4: Authentication with panic code
try:
db = Database(base_dir=test_base)
auth = Authentication(db)
auth.setup_master_password("TestPass123!", "PANIC123")
success, panic = auth.login("TestPass123!")
assert success == True and panic == False
success, panic = auth.login("WrongPass!")
assert success == False and panic == False
success, panic = auth.login("PANIC123")
assert success == False and panic == True
print("[PASS] test_authentication")
passed += 1
except Exception as e:
print(f"[FAIL] test_authentication: {e}")
failed += 1
# Test 5: Database metadata
try:
db = Database(base_dir=test_base)
db.set_metadata("test_key", "test_value")
val = db.get_metadata("test_key")
assert val == "test_value"
print("[PASS] test_database_metadata")
passed += 1
except Exception as e:
print(f"[FAIL] test_database_metadata: {e}")
failed += 1
# Test 6: Session lock clears encryption
try:
db = Database(base_dir=test_base)
sec = SecurityManager(db)
sec.setup("SessionTest123!", "Panic1234")
assert sec.is_active == True
sec.lock()
assert sec.is_locked == True
assert sec.encryption is None
print("[PASS] test_session_lock")
passed += 1
except Exception as e:
print(f"[FAIL] test_session_lock: {e}")
failed += 1
# Test 7: Panic mode wipes data
try:
db = Database(base_dir=test_base)
sec = SecurityManager(db)
sec.setup("PanicTest123!", "DESTROY")
sec.login("DESTROY")
assert sec.is_locked == True
assert sec.encryption is None
with db._connect() as conn:
count = conn.execute("SELECT COUNT(*) FROM passwords").fetchone()[0]
assert count == 0
print("[PASS] test_panic_mode")
passed += 1
except Exception as e:
print(f"[FAIL] test_panic_mode: {e}")
failed += 1
# Test 8: Export/Import roundtrip (no ID conflicts)
try:
db = Database(base_dir=test_base)
sec = SecurityManager(db)
sec.setup("ExportTest123!", "Panic9999")
pm = PasswordManager(db, sec.encryption)
pm.add_password("test_service", "test_user", "test_pass")
bundle = db.export_encrypted(sec.encryption)
test_base2 = test_base / "import_test"
db2 = Database(base_dir=test_base2)
sec2 = SecurityManager(db2)
sec2.setup("ExportTest123!", "Panic9999")
db2.import_encrypted(bundle, sec2.encryption)
pm2 = PasswordManager(db2, sec2.encryption)
passwords = pm2.get_passwords()
assert len(passwords) == 1
assert passwords[0]["service"] == "test_service"
print("[PASS] test_export_import")
passed += 1
except Exception as e:
print(f"[FAIL] test_export_import: {e}")
failed += 1
# Test 9: Self-destruct after max failed logins
try:
db = Database(base_dir=test_base)
sec = SecurityManager(db)
sec.setup("SelfDestruct123!", "Panic0000")
for i in range(Config.MAX_FAILED_LOGINS):
success, panic = sec.login("WrongPassword")
if i < Config.MAX_FAILED_LOGINS - 1:
assert panic == False
else:
assert panic == True
assert sec.is_locked == True
print("[PASS] test_self_destruct")
passed += 1
except Exception as e:
print(f"[FAIL] test_self_destruct: {e}")
failed += 1
# Test 10: Distance estimation
try:
db = Database(base_dir=test_base)
scanner = SignalScanner(db)
dist, conf = scanner._estimate_distance(-60, 2400)
assert dist > 0
assert conf in ["HIGH", "MEDIUM", "LOW", "UNKNOWN"]
print("[PASS] test_distance_estimation")
passed += 1
except Exception as e:
print(f"[FAIL] test_distance_estimation: {e}")
failed += 1
print("\n" + "=" * 50)
print(f"RESULTS: {passed} passed, {failed} failed")
print("=" * 50 + "\n")
return failed == 0
# ============================================================
# MAIN
# ============================================================
def main():
if len(sys.argv) > 1 and sys.argv[1] == "--test":
success = run_tests()
sys.exit(0 if success else 1)
print("""
+======================================+
| SovereignX Security Hub v5.0 |
| Portable Data & Security Tool |
+======================================+
""")
time.sleep(1)
try:
db = Database()
security = SecurityManager(db)
ui = TerminalUI(security, db)
ui.run()
except KeyboardInterrupt:
print("\n\nStopped by user")
except Exception as e:
logger.exception("Fatal error")
print(f"\nFatal error: {e}")
finally:
print("\nThank you for using SovereignX Security Hub")
if __name__ == "__main__":
main()
التعليقات (0)
سجّل الدخول للتعليق.