From 96b5f4e4703bfe9c05ca82fe24c359146c6b7095 Mon Sep 17 00:00:00 2001 From: Sidzeppelin95 <50958322+Sidzeppelin95@users.noreply.github.com> Date: Wed, 9 Sep 2026 23:29:50 +0530 Subject: [PATCH 1/2] Resolve wallet security review issues --- pishield/backend/app.py | 116 ++++++++++++++++------------ pishield/backend/config.py | 65 +++++++++++----- pishield/backend/security_engine.py | 101 ++++++++++++++++++++++-- pishield/backend/security_shared.py | 39 ++++++++++ pishield/backend/wallet_manager.py | 51 +++++++++++- 5 files changed, 292 insertions(+), 80 deletions(-) create mode 100644 pishield/backend/security_shared.py diff --git a/pishield/backend/app.py b/pishield/backend/app.py index 5128708..89fc4e3 100644 --- a/pishield/backend/app.py +++ b/pishield/backend/app.py @@ -1,4 +1,6 @@ +"""Flask application for the PiShield sandbox demo.""" from pathlib import Path + from flask import Flask, jsonify, request, send_from_directory from flask_cors import CORS @@ -10,61 +12,68 @@ from config import PI_APP_NAME, PI_APP_URL, PI_NETWORK, PI_SANDBOX, PI_SANDBOX_URL from security_engine import SecurityEngine from wallet_manager import build_device_fingerprint - + app = Flask(__name__) - CORS(app) - +CORS(app) + security_engine = SecurityEngine() FRONTEND_DIR = Path(__file__).resolve().parents[1] / "frontend" - @app.route("/") - def home(): +@app.route("/") +def home(): + return jsonify( + { + "app": PI_APP_NAME, + "status": "sandbox_running", + "pi_network": "sandbox" if PI_SANDBOX else PI_NETWORK, + "security": "enabled", + "app_url": PI_APP_URL, + "sandbox_url": PI_SANDBOX_URL, + } + ) + - return jsonify({ - "app": PI_APP_NAME, - "status": "sandbox_running", - "pi_network": "sandbox" if PI_SANDBOX else PI_NETWORK, - "security": "enabled", - "app_url": PI_APP_URL, - "sandbox_url": PI_SANDBOX_URL, - }) - @app.route("/app") def frontend_app(): return send_from_directory(FRONTEND_DIR, "index.html") + + @app.route("/frontend/") def frontend_asset(filename): return send_from_directory(FRONTEND_DIR, filename) - - - @app.route("/health") - def health(): - return jsonify({ - "server": "online", - "wallet_security": "active", - "sandbox": PI_SANDBOX, - }) - - + + +@app.route("/health") +def health(): + return jsonify( + {"server": "online", "wallet_security": "active", "sandbox": PI_SANDBOX} + ) + + @app.route("/pi/validate") def validate(): - return jsonify({ - "app": PI_APP_NAME, - "verified": True, - "network": "sandbox" if PI_SANDBOX else PI_NETWORK, - }) - + return jsonify( + { + "app": PI_APP_NAME, + "verified": True, + "network": "sandbox" if PI_SANDBOX else PI_NETWORK, + } + ) + + @app.route("/api/config") def frontend_config(): - return jsonify({ - "app": PI_APP_NAME, - "network": PI_NETWORK, - "sandbox": PI_SANDBOX, - "app_url": PI_APP_URL, - "sandbox_url": PI_SANDBOX_URL, - }) - + return jsonify( + { + "app": PI_APP_NAME, + "network": PI_NETWORK, + "sandbox": PI_SANDBOX, + "app_url": PI_APP_URL, + "sandbox_url": PI_SANDBOX_URL, + } + ) + @app.route("/api/mfa/challenge", methods=["POST"]) def create_mfa_challenge(): @@ -79,20 +88,27 @@ def fingerprint(): payload = request.get_json(silent=True) or {} username = payload.get("username", "pi_sandbox_user") user_agent = request.headers.get("User-Agent", "unknown-agent") - return jsonify({ - "username": username, - "device_fingerprint": build_device_fingerprint(user_agent, username), - }) + return jsonify( + { + "username": username, + "device_fingerprint": build_device_fingerprint(user_agent, username), + } + ) + @app.route("/api/wallet/rotate-passphrase", methods=["POST"]) def rotate_passphrase(): payload = request.get_json(silent=True) or {} pi_auth_uid = payload.get("pi_auth_uid") - if "pi_auth_uid" in payload and (not isinstance(pi_auth_uid, str) or not pi_auth_uid.strip()): - return jsonify({ - "error": "invalid_pi_auth_uid", - "message": "If provided, pi_auth_uid must be a non-empty string.", - }), 400 + if "pi_auth_uid" in payload and ( + not isinstance(pi_auth_uid, str) or not pi_auth_uid.strip() + ): + return jsonify( + { + "error": "invalid_pi_auth_uid", + "message": "If provided, pi_auth_uid must be a non-empty string.", + } + ), 400 result, status = security_engine.rotate_passphrase( username=payload.get("username", "pi_sandbox_user"), @@ -102,7 +118,7 @@ def rotate_passphrase(): device_fingerprint=payload.get("device_fingerprint", ""), mfa_challenge_id=payload.get("mfa_challenge_id"), pi_auth_uid=pi_auth_uid.strip() if isinstance(pi_auth_uid, str) else None, - ) + ) return jsonify(result), status @@ -112,4 +128,4 @@ def security_dashboard(): if __name__ == "__main__": -+ app.run(host="0.0.0.0", port=31415, debug=True) + app.run(host="0.0.0.0", port=31415, debug=True) diff --git a/pishield/backend/config.py b/pishield/backend/config.py index 281370f..893a829 100644 --- a/pishield/backend/config.py +++ b/pishield/backend/config.py @@ -1,33 +1,58 @@ - -""PiShield sandbox configuration. - -Values default to the Pi Developer Portal sandbox settings and can be -overridden with environment variables for local/ngrok development. -"" - +"""PiShield sandbox configuration.""" +from __future__ import annotations import os + from dotenv import load_dotenv + +load_dotenv() + + def _env_bool(name: str, default: bool) -> bool: value = os.getenv(name) if value is None: return default return value.strip().lower() in {"1", "true", "yes", "on"} - +class PiOSConfig: + """Configuration values for Pi Browser and wallet-security workflows.""" + + APP_NAME = os.getenv("PI_APP_NAME", "PiShield") + APP_VERSION = "1.0.0" + PIOS_COMPATIBLE = True + + PROD_APP_URL = os.getenv("PI_PROD_APP_URL", "https://pishield.pinet.com") + DEV_APP_URL = os.getenv("PI_APP_URL", "http://localhost:31415") + API_URL = os.getenv("PI_API_URL", "https://api.minepi.com") + PRIVACY_POLICY_URL = os.getenv( + "PRIVACY_POLICY_URL", "https://pishield.pinet.com/privacy-policy" + ) + TERMS_OF_SERVICE_URL = os.getenv( + "TERMS_OF_SERVICE_URL", "https://pishield.pinet.com/terms" + ) + SANDBOX_URL = os.getenv("PI_SANDBOX_URL", "https://sandbox.minepi.com") + + PI_BROWSER_REQUIRED = True + PI_SDK_ENABLED = True + PI_MAINNET_ENABLED = True + + ROTATION_DELAY_HOURS = 48 + RECOVERY_LOCK_HOURS = 24 + THREAT_SCORE_THRESHOLD = 70 + HIGH_RISK_THRESHOLD = 90 + MAX_RECOVERY_ATTEMPTS = 3 + + +# Backwards-compatible aliases used by the existing Flask app/config imports. +# TODO: migrate callers to PiOSConfig and remove these aliases. PI_SANDBOX = _env_bool("PI_SANDBOX", True) -PI_APP_NAME = os.getenv("PI_APP_NAME", "PiShield") +PI_APP_NAME = PiOSConfig.APP_NAME PI_API_KEY = os.getenv("PI_API_KEY", "YOUR_PI_API_KEY") PI_NETWORK = os.getenv("PI_NETWORK", "Pi Testnet") -PI_APP_URL = os.getenv("PI_APP_URL", "http://localhost:31415") -PI_API_URL = os.getenv("PI_API_URL", "https://api.minepi.com") -PI_SANDBOX_URL = os.getenv("PI_SANDBOX_URL", "https://sandbox.minepi.com") -PRIVACY_POLICY_URL = os.getenv( - "PRIVACY_POLICY_URL", - "https://pishield.pinet.com/privacy-policy", -) -TERMS_OF_SERVICE_URL = os.getenv( - "TERMS_OF_SERVICE_URL", - "https://pishield.pinet.com/terms", - ) +# Legacy Flask uses the development URL for the app. +PI_APP_URL = PiOSConfig.DEV_APP_URL +PI_API_URL = PiOSConfig.API_URL +PI_SANDBOX_URL = PiOSConfig.SANDBOX_URL +PRIVACY_POLICY_URL = PiOSConfig.PRIVACY_POLICY_URL +TERMS_OF_SERVICE_URL = PiOSConfig.TERMS_OF_SERVICE_URL diff --git a/pishield/backend/security_engine.py b/pishield/backend/security_engine.py index 50950be..d6f2d83 100644 --- a/pishield/backend/security_engine.py +++ b/pishield/backend/security_engine.py @@ -1,18 +1,25 @@ """Demo security primitives for PiShield. - -+This module intentionally keeps state in memory because the project is a -+sandbox demonstration. Production deployments should persist audit trails, -+store only salted password hashes, validate Pi auth tokens server-side, and -+separate sandbox/mainnet databases. + +This module intentionally keeps state in memory because the project is a +sandbox demonstration. Production deployments should persist audit trails, +store only salted password hashes, validate Pi auth tokens server-side, and +separate sandbox/mainnet databases. """ from __future__ import annotations from dataclasses import dataclass, field -from datetime import datetime, timezone -from hashlib import sha256 +from datetime import datetime, timedelta, timezone +import ipaddress from secrets import token_hex from typing import Any +try: + from .config import PiOSConfig + from .security_shared import ConnectionMetadata, SecurityUtils, Wallet, utc_now +except ImportError: # Supports running the backend as a script. + from config import PiOSConfig + from security_shared import ConnectionMetadata, SecurityUtils, Wallet, utc_now + @dataclass class RotationRecord: @@ -148,8 +155,86 @@ def _flag( @staticmethod def _hash(value: str) -> str: - return sha256(value.encode("utf-8")).hexdigest() + return SecurityUtils.hash_passphrase(value) @staticmethod def _now() -> str: return datetime.now(timezone.utc).isoformat() + + +@dataclass +class SecurityEvent: + """An event that may require analyst review.""" + + event_id: str + wallet_username: str + created_at: datetime = field(default_factory=utc_now) + status: str = "PENDING_REVIEW" + analyst_notes: str = "" + + +security_events_db: dict[str, SecurityEvent] = {} + + +class PiTrustAnalyzer: + """Calculate risk from explicit connection facts and not device-ID text.""" + + @staticmethod + def calculate_risk_score( + device_id: str, + connection: ConnectionMetadata | None = None, + ) -> int: + """Return a bounded risk score for the supplied connection metadata. + + ``device_id`` remains available for device-recognition logic, but it is + intentionally not parsed for network labels such as "vpn" or "tor". + """ + del device_id + connection = connection or ConnectionMetadata() + score = 0 + if connection.uses_vpn: + score += 20 + if connection.uses_tor: + score += 35 + + if connection.ip_address: + try: + ip_obj = ipaddress.ip_address(connection.ip_address) + except ValueError: + pass + else: + if ip_obj.is_private: + score -= 10 + return max(0, min(100, score)) + + +class PiSecurityEngine: + """Coordinates recovery responses without importing the wallet manager.""" + + def trigger_response(self, wallet: Wallet) -> None: + """Lock a wallet for the configured recovery period after an incident.""" + wallet.recovery_locked_until = utc_now() + timedelta( + hours=PiOSConfig.RECOVERY_LOCK_HOURS + ) + + +class PiSecurityReviewSystem: + """Analyst review workflow for flagged security events.""" + + @staticmethod + def review_event(event_id: str, suspicious: bool, notes: str) -> SecurityEvent: + event = security_events_db.get(event_id) + if event is None: + raise ValueError("Security event not found") + + event.analyst_notes = notes + if suspicious: + PiSecurityReviewSystem.take_action(event) + else: + event.status = "CLEARED" + return event + + @staticmethod + def take_action(event: SecurityEvent) -> None: + """Mark a reviewed event as confirmed phishing activity.""" + event.status = "CONFIRMED_PHISHING_ACTIVITY" diff --git a/pishield/backend/security_shared.py b/pishield/backend/security_shared.py new file mode 100644 index 0000000..88ca752 --- /dev/null +++ b/pishield/backend/security_shared.py @@ -0,0 +1,39 @@ +"""Shared security primitives used by wallet and security services.""" +from __future__ import annotations + +from dataclasses import dataclass, field +from datetime import datetime, timezone +from hashlib import sha256 + + +def utc_now() -> datetime: + """Return a timezone-aware timestamp in UTC.""" + return datetime.now(timezone.utc) + + +@dataclass +class Wallet: + """Shared wallet state used by authentication and recovery services.""" + + username: str + active_passphrase_hash: str + created_at: datetime = field(default_factory=utc_now) + revoked_passphrase_hashes: set[str] = field(default_factory=set) + recovery_locked_until: datetime | None = None + + +class SecurityUtils: + """Stateless helpers shared across security components.""" + + @staticmethod + def hash_passphrase(passphrase: str) -> str: + return sha256(passphrase.encode("utf-8")).hexdigest() + + +@dataclass(frozen=True) +class ConnectionMetadata: + """Connection facts supplied by the caller rather than encoded in an ID.""" + + ip_address: str | None = None + uses_vpn: bool = False + uses_tor: bool = False diff --git a/pishield/backend/wallet_manager.py b/pishield/backend/wallet_manager.py index e09fe79..1a829ca 100644 --- a/pishield/backend/wallet_manager.py +++ b/pishield/backend/wallet_manager.py @@ -1,8 +1,55 @@ -"""Wallet helpers for the PiShield sandbox demo.""" - +"""Wallet authentication and device helpers for the PiShield sandbox demo.""" from __future__ import annotations +from dataclasses import dataclass, field from hashlib import sha256 +from typing import Protocol + +try: + from .security_shared import SecurityUtils, Wallet, utc_now +except ImportError: # Supports running the backend as a script. + from security_shared import SecurityUtils, Wallet, utc_now + + +class RecoveryResponseHandler(Protocol): + """The narrow interface wallet authentication needs from a security service.""" + + def trigger_response(self, wallet: "Wallet") -> None: ... + + +@dataclass +class PiWalletManager: + """In-memory wallet store with lock-aware passphrase authentication.""" + + wallets: dict[str, Wallet] = field(default_factory=dict) + response_handler: RecoveryResponseHandler | None = None + + def create_wallet(self, username: str, passphrase: str) -> Wallet: + wallet = Wallet( + username=username, + active_passphrase_hash=SecurityUtils.hash_passphrase(passphrase), + ) + self.wallets[username] = wallet + return wallet + + def authenticate(self, username: str, entered_passphrase: str) -> bool: + wallet = self.wallets.get(username) + if wallet is None: + raise ValueError("Wallet not found") + + # A recovery lock applies to every authentication attempt, including a + # revoked passphrase, until its timezone-aware UTC expiry has passed. + if wallet.recovery_locked_until and wallet.recovery_locked_until > utc_now(): + return False + + entered_hash = SecurityUtils.hash_passphrase(entered_passphrase) + if entered_hash == wallet.active_passphrase_hash: + return True + + if entered_hash in wallet.revoked_passphrase_hashes: + if self.response_handler is not None: + self.response_handler.trigger_response(wallet) + return False def build_device_fingerprint(user_agent: str, pi_username: str) -> str: From 663d6a0aad8b7cfe6455e72a65aa2924adfa3d4e Mon Sep 17 00:00:00 2001 From: Sidzeppelin95 <50958322+Sidzeppelin95@users.noreply.github.com> Date: Thu, 10 Sep 2026 20:22:47 +0530 Subject: [PATCH 2/2] Wire sandbox recovery flow and add regressions --- README.md | 41 ++++++++++++ pishield/backend/app.py | 40 ++++++++++-- pishield/backend/security_engine.py | 22 ++++++- pishield/backend/security_shared.py | 13 ++++ pytest.ini | 3 + tests/test_app_integration.py | 95 ++++++++++++++++++++++++++++ tests/test_security_engine.py | 62 +++++++++++++++++++ tests/test_wallet_recovery.py | 96 +++++++++++++++++++++++++++++ 8 files changed, 365 insertions(+), 7 deletions(-) create mode 100644 pytest.ini create mode 100644 tests/test_app_integration.py create mode 100644 tests/test_security_engine.py create mode 100644 tests/test_wallet_recovery.py diff --git a/README.md b/README.md index d9efaff..fe8dc1d 100644 --- a/README.md +++ b/README.md @@ -47,6 +47,7 @@ pishield/ │ ├── app.py │ ├── config.py │ ├── security_engine.py +│ ├── security_shared.py │ └── wallet_manager.py │ ├── frontend/ @@ -54,6 +55,7 @@ pishield/ │ ├── app.js │ └── styles.css │ +├── tests/ ├── requirements.txt └── README.md ``` @@ -270,3 +272,42 @@ Testing includes: ## Disclaimer This repository is a research and development prototype intended to demonstrate secure wallet recovery concepts and passphrase rotation workflows. It is not affiliated with or endorsed by the Pi Core Team and should not be used in production without comprehensive security review, testing, and integration with the official Pi platform. + +--- + +## Sandbox security architecture + +PiShield is a **sandbox demonstration**. It does not connect to Pi Wallets, change +real Pi Wallet credentials, transmit credentials to third parties, or represent a +production credential provider. The demo-only credential hashes stay in process +memory and are never returned by the API. + +The Flask application keeps general MFA, rotation, audit, and dashboard behavior +in `SecurityEngine`. Its sandbox authentication path is intentionally separate: +`PiWalletManager` decides whether a demo credential is active, revoked, or +invalid. It receives `PiSecurityEngine` through the narrow +`RecoveryResponseHandler` interface. A revoked demo credential is denied and +causes a temporary recovery lock; it records a security signal, not a conclusion +that a person is a scammer or that phishing is confirmed. + +Wallet timestamps are normalized to timezone-aware UTC at the `Wallet` boundary. +While a recovery lock is active, all authentication attempts—including active, +revoked, and invalid demo credentials—are denied. Expired locks do not prevent a +valid active demo credential from being accepted. + +`PiTrustAnalyzer` is advisory only. It accepts explicit connection metadata +(`uses_vpn`, `uses_tor`, and `ip_address`) rather than inferring network behavior +from a device ID. VPN/Tor use, IP addresses, and fingerprints do not independently +identify a user as malicious. Confirmed phishing activity is assigned only by the +separate analyst review workflow. + +### Validation + +Run the sandbox regression suite from the repository root: + +```bash +python -m compileall -q pishield +pytest -q +python -m pytest -q +git diff --check +``` diff --git a/pishield/backend/app.py b/pishield/backend/app.py index 89fc4e3..d0fe342 100644 --- a/pishield/backend/app.py +++ b/pishield/backend/app.py @@ -6,17 +6,19 @@ if __package__: from .config import PI_APP_NAME, PI_APP_URL, PI_NETWORK, PI_SANDBOX, PI_SANDBOX_URL - from .security_engine import SecurityEngine - from .wallet_manager import build_device_fingerprint + from .security_engine import PiSecurityEngine, SecurityEngine + from .wallet_manager import PiWalletManager, build_device_fingerprint else: from config import PI_APP_NAME, PI_APP_URL, PI_NETWORK, PI_SANDBOX, PI_SANDBOX_URL - from security_engine import SecurityEngine - from wallet_manager import build_device_fingerprint + from security_engine import PiSecurityEngine, SecurityEngine + from wallet_manager import PiWalletManager, build_device_fingerprint app = Flask(__name__) CORS(app) security_engine = SecurityEngine() +pi_security_engine = PiSecurityEngine(general_security=security_engine) +wallet_manager = PiWalletManager(response_handler=pi_security_engine) FRONTEND_DIR = Path(__file__).resolve().parents[1] / "frontend" @@ -80,7 +82,35 @@ def create_mfa_challenge(): payload = request.get_json(silent=True) or {} username = payload.get("username", "pi_sandbox_user") pi_auth_uid = payload.get("pi_auth_uid") - return jsonify(security_engine.create_mfa_challenge(username, pi_auth_uid=pi_auth_uid)) + if "pi_auth_uid" in payload and ( + not isinstance(pi_auth_uid, str) or not pi_auth_uid.strip() + ): + return jsonify({"error": "invalid_pi_auth_uid"}), 400 + return jsonify( + security_engine.create_mfa_challenge( + username, + pi_auth_uid=pi_auth_uid.strip() if isinstance(pi_auth_uid, str) else None, + ) + ) + + +@app.route("/api/wallet/authenticate", methods=["POST"]) +def authenticate_wallet(): + """Authenticate a sandbox-only demo credential without exposing it.""" + payload = request.get_json(silent=True) or {} + username = payload.get("username", "pi_sandbox_user") + demo_credential = payload.get("demo_credential") + if not isinstance(username, str) or not isinstance(demo_credential, str): + return jsonify({"error": "username and demo_credential are required"}), 400 + + try: + authenticated = wallet_manager.authenticate(username, demo_credential) + except ValueError: + return jsonify({"error": "wallet not found"}), 404 + + return jsonify({"authenticated": authenticated, "sandbox": PI_SANDBOX}), ( + 200 if authenticated else 403 + ) @app.route("/api/wallet/fingerprint", methods=["POST"]) diff --git a/pishield/backend/security_engine.py b/pishield/backend/security_engine.py index d6f2d83..0fbc486 100644 --- a/pishield/backend/security_engine.py +++ b/pishield/backend/security_engine.py @@ -135,6 +135,16 @@ def dashboard(self) -> dict[str, Any]: "mfa_challenges": list(self.mfa_challenges.values())[-10:], } + def record_security_event( + self, + username: str, + reason: str, + *, + device_fingerprint: str = "recovery-service", + ) -> None: + """Record a sandbox security signal without assigning user intent.""" + self._flag(username, device_fingerprint, reason) + def _flag( self, username: str, @@ -208,14 +218,22 @@ def calculate_risk_score( return max(0, min(100, score)) +@dataclass class PiSecurityEngine: - """Coordinates recovery responses without importing the wallet manager.""" + """Coordinates temporary recovery locks through a narrow wallet contract.""" + + general_security: SecurityEngine | None = None def trigger_response(self, wallet: Wallet) -> None: - """Lock a wallet for the configured recovery period after an incident.""" + """Temporarily lock a wallet after a revoked demo credential is used.""" wallet.recovery_locked_until = utc_now() + timedelta( hours=PiOSConfig.RECOVERY_LOCK_HOURS ) + if self.general_security is not None: + self.general_security.record_security_event( + wallet.username, + "revoked_demo_credential_attempt", + ) class PiSecurityReviewSystem: diff --git a/pishield/backend/security_shared.py b/pishield/backend/security_shared.py index 88ca752..7034660 100644 --- a/pishield/backend/security_shared.py +++ b/pishield/backend/security_shared.py @@ -11,6 +11,15 @@ def utc_now() -> datetime: return datetime.now(timezone.utc) +def normalize_utc(value: datetime | None) -> datetime | None: + """Normalize a persisted timestamp to an aware UTC datetime.""" + if value is None: + return None + if value.tzinfo is None: + return value.replace(tzinfo=timezone.utc) + return value.astimezone(timezone.utc) + + @dataclass class Wallet: """Shared wallet state used by authentication and recovery services.""" @@ -21,6 +30,10 @@ class Wallet: revoked_passphrase_hashes: set[str] = field(default_factory=set) recovery_locked_until: datetime | None = None + def __post_init__(self) -> None: + self.created_at = normalize_utc(self.created_at) + self.recovery_locked_until = normalize_utc(self.recovery_locked_until) + class SecurityUtils: """Stateless helpers shared across security components.""" diff --git a/pytest.ini b/pytest.ini new file mode 100644 index 0000000..4584de7 --- /dev/null +++ b/pytest.ini @@ -0,0 +1,3 @@ +[pytest] +testpaths = tests +pythonpath = . diff --git a/tests/test_app_integration.py b/tests/test_app_integration.py new file mode 100644 index 0000000..51bffcc --- /dev/null +++ b/tests/test_app_integration.py @@ -0,0 +1,95 @@ +from datetime import timezone + +import pytest + +from pishield.backend import app as app_module +from pishield.backend.security_engine import PiSecurityEngine +from pishield.backend.security_shared import SecurityUtils, utc_now +from pishield.backend.wallet_manager import PiWalletManager + + +@pytest.fixture(autouse=True) +def reset_application_state(): + app_module.wallet_manager.wallets.clear() + app_module.security_engine.rotations.clear() + app_module.security_engine.revoked_passphrases.clear() + app_module.security_engine.suspicious_events.clear() + app_module.security_engine.mfa_challenges.clear() + yield + + +@pytest.fixture +def client(): + return app_module.app.test_client() + + +def test_flask_endpoints_preserve_sandbox_behavior(client): + assert client.get("/").status_code == 200 + assert client.get("/health").status_code == 200 + assert client.get("/pi/validate").status_code == 200 + assert client.get("/api/config").status_code == 200 + assert client.post("/api/mfa/challenge", json={"username": "sandbox-user"}).status_code == 200 + assert client.post("/api/wallet/fingerprint", json={"username": "sandbox-user"}).status_code == 200 + assert client.post("/api/wallet/rotate-passphrase", json={}).status_code == 400 + assert client.get("/api/security/dashboard").status_code == 200 + + +@pytest.mark.parametrize("pi_auth_uid", ["", 42]) +def test_rotate_rejects_empty_or_invalid_pi_auth_uid(client, pi_auth_uid): + response = client.post("/api/wallet/rotate-passphrase", json={"pi_auth_uid": pi_auth_uid}) + assert response.status_code == 400 + + +def test_rotate_accepts_valid_pi_auth_uid(client): + response = client.post( + "/api/wallet/rotate-passphrase", + json={ + "username": "sandbox-user", + "current_passphrase": "old-demo-passphrase", + "new_passphrase": "new-demo-passphrase", + "biometric_confirmed": True, + "pi_auth_uid": "uid-123", + }, + ) + assert response.status_code == 200 + assert response.get_json()["pi_auth_uid"] == "uid-123" + + +def test_application_wires_revoked_credential_to_recovery_lock(client): + assert isinstance(app_module.pi_security_engine, PiSecurityEngine) + assert isinstance(app_module.wallet_manager, PiWalletManager) + assert app_module.wallet_manager.response_handler is app_module.pi_security_engine + + wallet = app_module.wallet_manager.create_wallet("sandbox-user", "active-demo") + wallet.revoked_passphrase_hashes.add(SecurityUtils.hash_passphrase("revoked-demo")) + response = client.post( + "/api/wallet/authenticate", + json={"username": "sandbox-user", "demo_credential": "revoked-demo"}, + ) + + assert response.status_code == 403 + assert response.get_json()["authenticated"] is False + assert wallet.recovery_locked_until > utc_now() + assert wallet.recovery_locked_until.tzinfo is timezone.utc + assert app_module.security_engine.suspicious_events[-1]["reason"] == "revoked_demo_credential_attempt" + + +def test_authentication_endpoint_blocks_active_credential_while_locked(client): + wallet = app_module.wallet_manager.create_wallet("sandbox-user", "active-demo") + wallet.recovery_locked_until = utc_now().replace(year=utc_now().year + 1) + response = client.post( + "/api/wallet/authenticate", + json={"username": "sandbox-user", "demo_credential": "active-demo"}, + ) + assert response.status_code == 403 + +@pytest.mark.parametrize("pi_auth_uid", ["", 42]) +def test_mfa_challenge_rejects_empty_or_invalid_pi_auth_uid(client, pi_auth_uid): + response = client.post("/api/mfa/challenge", json={"pi_auth_uid": pi_auth_uid}) + assert response.status_code == 400 + + +def test_mfa_challenge_accepts_valid_pi_auth_uid(client): + response = client.post("/api/mfa/challenge", json={"pi_auth_uid": "uid-123"}) + assert response.status_code == 200 + assert response.get_json()["pi_auth_uid"] == "uid-123" diff --git a/tests/test_security_engine.py b/tests/test_security_engine.py new file mode 100644 index 0000000..704fb4e --- /dev/null +++ b/tests/test_security_engine.py @@ -0,0 +1,62 @@ +import pytest + +from pishield.backend.security_engine import PiTrustAnalyzer, SecurityEngine +from pishield.backend.security_shared import ConnectionMetadata +from pishield.backend.wallet_manager import build_device_fingerprint + + +@pytest.mark.parametrize( + ("metadata", "expected"), + [ + (ConnectionMetadata(ip_address="10.0.0.1"), 0), + (ConnectionMetadata(ip_address="fd00::1"), 0), + (ConnectionMetadata(ip_address="8.8.8.8"), 0), + (ConnectionMetadata(ip_address="invalid"), 0), + (ConnectionMetadata(uses_vpn=True), 20), + (ConnectionMetadata(uses_tor=True), 35), + (ConnectionMetadata(uses_vpn=True, uses_tor=True, ip_address="172.16.0.1"), 45), + ], +) +def test_connection_metadata_drives_bounded_risk_score(metadata, expected): + score = PiTrustAnalyzer.calculate_risk_score("tor-in-device-id-is-ignored", metadata) + assert score == expected + assert 0 <= score <= 100 + + +def test_mfa_challenge_validation_and_pi_auth_uid_binding(): + engine = SecurityEngine() + challenge = engine.create_mfa_challenge("sandbox-user", "uid-1") + result, status = engine.rotate_passphrase( + username="sandbox-user", current_passphrase="old-passphrase", new_passphrase="new-passphrase-123", + biometric_confirmed=True, device_fingerprint="device", mfa_challenge_id=challenge["challenge_id"], pi_auth_uid="uid-1" + ) + assert status == 200 and result["status"] == "passphrase_rotated" + + result, status = engine.rotate_passphrase( + username="sandbox-user", current_passphrase="another-old", new_passphrase="another-new-123", + biometric_confirmed=True, device_fingerprint="device", mfa_challenge_id="missing", pi_auth_uid="uid-1" + ) + assert status == 403 and result["error"] == "valid MFA challenge is required" + + challenge = engine.create_mfa_challenge("sandbox-user", "uid-2") + result, status = engine.rotate_passphrase( + username="sandbox-user", current_passphrase="third-old", new_passphrase="third-new-passphrase", + biometric_confirmed=True, device_fingerprint="device", mfa_challenge_id=challenge["challenge_id"], pi_auth_uid="wrong" + ) + assert status == 403 and result["error"] == "pi_auth_mismatch" + + +def test_device_fingerprint_is_deterministic_and_does_not_expose_user_agent(): + fingerprint = build_device_fingerprint("Mozilla/Test", "sandbox-user") + assert fingerprint == build_device_fingerprint("Mozilla/Test", "sandbox-user") + assert "Mozilla/Test" not in fingerprint + + +def test_config_aliases_are_driven_by_pios_config(): + from pishield.backend import config + + assert config.PI_SANDBOX is True + assert config.PI_APP_NAME == config.PiOSConfig.APP_NAME + assert config.PI_APP_URL == config.PiOSConfig.DEV_APP_URL + assert config.PI_API_URL == config.PiOSConfig.API_URL + assert config.PI_SANDBOX_URL == config.PiOSConfig.SANDBOX_URL diff --git a/tests/test_wallet_recovery.py b/tests/test_wallet_recovery.py new file mode 100644 index 0000000..4e7edfb --- /dev/null +++ b/tests/test_wallet_recovery.py @@ -0,0 +1,96 @@ +from datetime import datetime, timedelta, timezone + +import pytest + +from pishield.backend.config import PiOSConfig +from pishield.backend.security_engine import PiSecurityEngine +from pishield.backend.security_shared import SecurityUtils, Wallet, utc_now +from pishield.backend.wallet_manager import PiWalletManager + + +class RecordingHandler: + def __init__(self): + self.wallets = [] + + def trigger_response(self, wallet): + self.wallets.append(wallet) + + +def make_wallet(**kwargs): + return Wallet( + username="sandbox-user", + active_passphrase_hash=SecurityUtils.hash_passphrase("active-demo"), + **kwargs, + ) + + +def test_wallet_timestamps_are_aware_utc_and_normalize_naive_values(): + naive = datetime(2026, 1, 1, 12, 0) + wallet = make_wallet(created_at=naive, recovery_locked_until=naive) + assert wallet.created_at.tzinfo is timezone.utc + assert wallet.recovery_locked_until.tzinfo is timezone.utc + + +def test_aware_non_utc_lock_is_converted_to_utc(): + eastern = timezone(timedelta(hours=-4)) + lock = datetime(2026, 1, 1, 12, 0, tzinfo=eastern) + wallet = make_wallet(recovery_locked_until=lock) + assert wallet.recovery_locked_until == datetime(2026, 1, 1, 16, 0, tzinfo=timezone.utc) + + +def test_naive_lock_does_not_raise_and_blocks_authentication(): + manager = PiWalletManager() + wallet = make_wallet(recovery_locked_until=datetime.now() + timedelta(hours=1)) + manager.wallets[wallet.username] = wallet + assert manager.authenticate(wallet.username, "active-demo") is False + + +def test_expired_lock_allows_active_credential(): + manager = PiWalletManager() + wallet = make_wallet(recovery_locked_until=utc_now() - timedelta(seconds=1)) + manager.wallets[wallet.username] = wallet + assert manager.authenticate(wallet.username, "active-demo") is True + + +@pytest.mark.parametrize("credential", ["active-demo", "revoked-demo", "invalid-demo"]) +def test_future_lock_blocks_every_credential_type(credential): + manager = PiWalletManager() + wallet = make_wallet(recovery_locked_until=utc_now() + timedelta(hours=1)) + wallet.revoked_passphrase_hashes.add(SecurityUtils.hash_passphrase("revoked-demo")) + manager.wallets[wallet.username] = wallet + assert manager.authenticate(wallet.username, credential) is False + + +def test_active_and_revoked_credential_behavior_with_handler(): + handler = RecordingHandler() + manager = PiWalletManager(response_handler=handler) + wallet = manager.create_wallet("sandbox-user", "active-demo") + wallet.revoked_passphrase_hashes.add(SecurityUtils.hash_passphrase("revoked-demo")) + assert manager.authenticate(wallet.username, "active-demo") is True + assert manager.authenticate(wallet.username, "revoked-demo") is False + assert handler.wallets == [wallet] + + +def test_revoked_credential_without_handler_fails_safely(): + manager = PiWalletManager() + wallet = manager.create_wallet("sandbox-user", "active-demo") + wallet.revoked_passphrase_hashes.add(SecurityUtils.hash_passphrase("revoked-demo")) + assert manager.authenticate(wallet.username, "revoked-demo") is False + + +def test_pi_security_engine_applies_aware_configured_lock(): + wallet = make_wallet() + before = utc_now() + PiSecurityEngine().trigger_response(wallet) + assert wallet.recovery_locked_until.tzinfo is timezone.utc + expected = before + timedelta(hours=PiOSConfig.RECOVERY_LOCK_HOURS) + assert abs((wallet.recovery_locked_until - expected).total_seconds()) < 1 + + +def test_injected_pi_security_engine_locks_wallet_after_revoked_credential(): + engine = PiSecurityEngine() + manager = PiWalletManager(response_handler=engine) + wallet = manager.create_wallet("sandbox-user", "active-demo") + wallet.revoked_passphrase_hashes.add(SecurityUtils.hash_passphrase("revoked-demo")) + assert manager.authenticate(wallet.username, "revoked-demo") is False + assert wallet.recovery_locked_until > utc_now()