diff --git a/pishield/backend/config.py b/pishield/backend/config.py index 63332c6..d2e8c13 100644 --- a/pishield/backend/config.py +++ b/pishield/backend/config.py @@ -1,19 +1,36 @@ -PI_SANDBOX = True +"""PiShield PiOS-compatible configuration.""" -PI_APP_NAME = "PiShield" -PI_API_KEY = "YOUR_PI_API_KEY" +class PiOSConfig: + """Configuration values for Pi Browser and wallet-security workflows.""" -PI_NETWORK = "Pi Testnet" + APP_NAME = "PiShield" + APP_VERSION = "1.0.0" + PIOS_COMPATIBLE = True -PI_APP_URL = "http://localhost:31415" + APP_URL = "https://pishield.pinet.com" + DEV_URL = "http://localhost:31415" + API_URL = "https://api.pishield.pinet.com" + PRIVACY_POLICY_URL = "https://pishield.pinet.com/privacy-policy" + TERMS_OF_SERVICE_URL = "https://pishield.pinet.com/terms" -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 -PRIVACY_POLICY_URL = ( - "https://pishield.pinet.com/privacy-policy" -) -TERMS_OF_SERVICE_URL = ( - "https://pishield.pinet.com/terms" -) +# Backwards-compatible aliases used by the existing Flask app/config imports. +PI_SANDBOX = True +PI_APP_NAME = PiOSConfig.APP_NAME +PI_API_KEY = "YOUR_PI_API_KEY" +PI_NETWORK = "Pi Testnet" +PI_APP_URL = PiOSConfig.DEV_URL +PI_SANDBOX_URL = "https://sandbox.minepi.com" +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 8b13789..12af4bd 100644 --- a/pishield/backend/security_engine.py +++ b/pishield/backend/security_engine.py @@ -1 +1,169 @@ +"""PiShield trust analysis and revoked-passphrase response engine. +Old passphrases never provide wallet access. Revoked passphrase attempts are +used only to analyze intent, evaluate trust signals, assist secure recovery, +flag suspicious behavior, and escalate questionable attempts. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from datetime import datetime, timedelta +from typing import TYPE_CHECKING, Dict, Optional + +from config import PiOSConfig +from wallet_manager import SecurityUtils + +if TYPE_CHECKING: + from wallet_manager import PiWallet + + +@dataclass +class SecurityEvent: + """Audit record for wallet recovery and suspicious authentication events.""" + + event_id: str + wallet_id: str + timestamp: datetime + ip_address: str + device_id: str + risk_score: int + classification: str + status: str + recovery_attempt: bool + analyst_notes: Optional[str] = None + + +security_events_db: Dict[str, SecurityEvent] = {} + + +class PiTrustAnalyzer: + """Evaluates device and network signals for revoked-passphrase attempts.""" + + @staticmethod + def is_trusted_device(wallet: "PiWallet", device_id: str) -> bool: + return any(device.device_id == device_id for device in wallet.trusted_devices) + + @staticmethod + def is_trusted_ip(wallet: "PiWallet", ip_address: str) -> bool: + return ip_address in wallet.trusted_ips + + @staticmethod + def get_device(wallet: "PiWallet", device_id: str): + for device in wallet.trusted_devices: + if device.device_id == device_id: + return device + return None + + @staticmethod + def calculate_risk_score(wallet: "PiWallet", ip_address: str, device_id: str) -> int: + score = 0 + trusted_device = PiTrustAnalyzer.is_trusted_device(wallet, device_id) + trusted_ip = PiTrustAnalyzer.is_trusted_ip(wallet, ip_address) + + if not trusted_device: + score += 40 + + if not trusted_ip: + score += 30 + + device = PiTrustAnalyzer.get_device(wallet, device_id) + if device: + if not device.pi_browser_verified: + score += 20 + if not device.biometric_enabled: + score += 15 + + lowered_device_id = device_id.lower() + if "vpn" in lowered_device_id: + score += 20 + if "tor" in lowered_device_id: + score += 35 + + if ip_address.startswith("192.168"): + score -= 10 + + return max(0, min(score, 100)) + + @staticmethod + def classify_attempt(risk_score: int, trusted_device: bool, trusted_ip: bool) -> str: + if trusted_device and trusted_ip and risk_score < 40: + return "LIKELY_GENUINE_OWNER_RECOVERY" + + if risk_score < PiOSConfig.THREAT_SCORE_THRESHOLD: + return "VERIFICATION_REQUIRED" + + return "LIKELY_PHISHING_ACTOR" + + +class PiSecurityEngine: + """Creates security events and triggers recovery or escalation responses.""" + + @staticmethod + def handle_old_passphrase_attempt(wallet: "PiWallet", ip_address: str, device_id: str) -> SecurityEvent: + trusted_device = PiTrustAnalyzer.is_trusted_device(wallet, device_id) + trusted_ip = PiTrustAnalyzer.is_trusted_ip(wallet, ip_address) + risk_score = PiTrustAnalyzer.calculate_risk_score(wallet, ip_address, device_id) + classification = PiTrustAnalyzer.classify_attempt( + risk_score, + trusted_device, + trusted_ip, + ) + + event = SecurityEvent( + event_id=SecurityUtils.generate_id(), + wallet_id=wallet.wallet_id, + timestamp=SecurityUtils.utc_now(), + ip_address=ip_address, + device_id=device_id, + risk_score=risk_score, + classification=classification, + status="FLAGGED", + recovery_attempt=True, + ) + security_events_db[event.event_id] = event + + PiSecurityEngine.trigger_response(wallet, event) + return event + + @staticmethod + def trigger_response(wallet: "PiWallet", event: SecurityEvent) -> None: + if event.classification == "LIKELY_GENUINE_OWNER_RECOVERY": + event.status = "RECOVERY_VERIFICATION_REQUIRED" + return + + if event.classification == "VERIFICATION_REQUIRED": + event.status = "SECURITY_REVIEW_QUEUED" + return + + if event.classification == "LIKELY_PHISHING_ACTOR": + wallet.suspicious_attempt_count += 1 + wallet.recovery_locked_until = SecurityUtils.utc_now() + timedelta( + hours=PiOSConfig.RECOVERY_LOCK_HOURS, + ) + event.status = "ESCALATED" + + +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 not event: + raise ValueError("Security event not found") + + event.analyst_notes = notes + if suspicious: + event.status = "CONFIRMED_PHISHING_ACTIVITY" + PiSecurityReviewSystem.take_action(event) + else: + event.status = "FALSE_POSITIVE" + + return event + + @staticmethod + def take_action(event: SecurityEvent) -> None: + # Placeholder for integrations with monitoring, device blacklists, and + # Pi ecosystem alerting systems. + event.status = "CONFIRMED_PHISHING_ACTIVITY" diff --git a/pishield/backend/wallet_manager.py b/pishield/backend/wallet_manager.py index 8b13789..d77cef9 100644 --- a/pishield/backend/wallet_manager.py +++ b/pishield/backend/wallet_manager.py @@ -1 +1,124 @@ +"""PiShield wallet models and passphrase rotation/authentication workflows.""" +from __future__ import annotations + +from dataclasses import dataclass, field +from datetime import datetime +from typing import Dict, List, Optional +import hashlib +import uuid + + +class SecurityUtils: + """Shared helpers for wallet-security workflows.""" + + @staticmethod + def hash_passphrase(passphrase: str) -> str: + """Hash a passphrase for demo storage. + + Production deployments should replace SHA-256 with a password hashing + algorithm such as Argon2id and use per-passphrase salts. + """ + + return hashlib.sha256(passphrase.encode()).hexdigest() + + @staticmethod + def generate_id() -> str: + return str(uuid.uuid4()) + + @staticmethod + def utc_now() -> datetime: + return datetime.utcnow() + + +@dataclass +class PiDeviceProfile: + """Known Pi Browser device profile for trust analysis.""" + + device_id: str + device_name: str + operating_system: str + pi_browser_verified: bool + biometric_enabled: bool + trusted_since: datetime + last_active: datetime + + +@dataclass +class PiWallet: + """Wallet state tracked by PiShield.""" + + wallet_id: str + pi_user_id: str + active_passphrase_hash: str + created_at: datetime = field(default_factory=SecurityUtils.utc_now) + trusted_ips: List[str] = field(default_factory=list) + trusted_devices: List[PiDeviceProfile] = field(default_factory=list) + revoked_passphrase_hashes: List[str] = field(default_factory=list) + recovery_locked_until: Optional[datetime] = None + suspicious_attempt_count: int = 0 + + +wallet_db: Dict[str, PiWallet] = {} + + +class PiWalletManager: + """Creates wallets, rotates passphrases, and authenticates attempts.""" + + @staticmethod + def create_wallet( + pi_user_id: str, + passphrase: str, + trusted_ip: str, + device_profile: PiDeviceProfile, + ) -> PiWallet: + wallet = PiWallet( + wallet_id=SecurityUtils.generate_id(), + pi_user_id=pi_user_id, + active_passphrase_hash=SecurityUtils.hash_passphrase(passphrase), + ) + wallet.trusted_ips.append(trusted_ip) + wallet.trusted_devices.append(device_profile) + wallet_db[wallet.wallet_id] = wallet + return wallet + + @staticmethod + def rotate_passphrase(wallet_id: str, current_passphrase: str, new_passphrase: str) -> PiWallet: + wallet = wallet_db.get(wallet_id) + if not wallet: + raise ValueError("Wallet not found") + + current_hash = SecurityUtils.hash_passphrase(current_passphrase) + if current_hash != wallet.active_passphrase_hash: + raise ValueError("Invalid active passphrase") + + wallet.revoked_passphrase_hashes.append(current_hash) + wallet.active_passphrase_hash = SecurityUtils.hash_passphrase(new_passphrase) + return wallet + + @staticmethod + def authenticate( + wallet_id: str, + entered_passphrase: str, + ip_address: str, + device_id: str, + ) -> bool: + wallet = wallet_db.get(wallet_id) + if not wallet: + raise ValueError("Wallet not found") + + entered_hash = SecurityUtils.hash_passphrase(entered_passphrase) + if entered_hash == wallet.active_passphrase_hash: + return True + + if entered_hash in wallet.revoked_passphrase_hashes: + from security_engine import PiSecurityEngine + + PiSecurityEngine.handle_old_passphrase_attempt( + wallet=wallet, + ip_address=ip_address, + device_id=device_id, + ) + return False + + return False