Add PiOS wallet security architecture

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

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"

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