//! Comprehensive tests for WireGuard road-warrior data-plane, cryptokey routing, //! endpoint resolution, telemetry ingestion, and reconciliation invariants. use axum::body::Body; use axum::http::{Request, StatusCode}; use chrono::Utc; use ipnet::IpNet; use nx9_wg_api::reconciliation::ReconciliationEngine; use nx9_wg_api::routes::build_api_router; use nx9_wg_api::state::AppState; use nx9_wg_core::crypto::generate_keypair; use nx9_wg_core::types::wireguard::{Interface, Peer, PeerProfile, PeerState, PeerType}; use nx9_wg_db::Store; use nx9_wg_network::{NetworkEngine, SimulatedNetworkEngine}; use nx9_wireguard::{ ClientConfigBuilder, LiveInterfaceStats, LivePeerStats, SimulatedWireGuardEngine, WireGuardEngine, }; use std::str::FromStr; use std::sync::Arc; use tower::ServiceExt; use uuid::Uuid; async fn setup_test_context() -> (AppState, Interface, Peer, String) { let store = Store::connect_in_memory().await.unwrap(); store.migrate().await.unwrap(); let now = Utc::now().naive_utc(); let hash = nx9_wg_core::crypto::hash_password("testadminpass123").unwrap(); store.create_admin("admin", &hash).await.unwrap(); let session = nx9_wg_core::types::auth::Session { id: "test-dataplane-session-id".to_string(), admin_id: 1, created_at: now, expires_at: now + chrono::Duration::hours(24), last_seen_at: Some(now), ip_address: Some("127.0.0.1".to_string()), user_agent: Some("test-agent".to_string()), }; store.create_session(&session).await.unwrap(); let (srv_priv, srv_pub) = generate_keypair(); let (peer_priv, peer_pub) = generate_keypair(); let interface = Interface { id: Uuid::new_v4(), name: "wg0".to_string(), private_key: srv_priv, public_key: srv_pub, listen_port: 51820, address_v4: IpNet::from_str("10.100.0.1/24").unwrap(), address_v6: None, mtu: Some(1420), dns: Some("1.1.1.1, 1.0.0.1".to_string()), enabled: true, pre_up: None, post_up: None, pre_down: None, post_down: None, created_at: now, updated_at: now, }; store.create_interface(&interface).await.unwrap(); let peer = Peer { id: Uuid::new_v4(), interface_id: interface.id, name: "Mobile".to_string(), peer_type: PeerType::RoadWarrior, state: PeerState::Active, public_key: peer_pub, private_key: Some(peer_priv), preshared_key: None, endpoint: None, allowed_ips: "0.0.0.0/0, ::/0".to_string(), server_allowed_ips: None, address_v4: Some(IpNet::from_str("10.100.0.9/32").unwrap()), address_v6: None, dns: Some("1.1.1.1, 1.0.0.1".to_string()), mtu: Some(1420), persistent_keepalive: Some(25), profile: PeerProfile::FullTunnel, expires_at: None, last_handshake_at: None, created_at: now, updated_at: now, }; store.create_peer(&peer).await.unwrap(); let state = AppState::new(store); (state, interface, peer, session.id) } #[tokio::test] async fn test_road_warrior_server_allowed_ips_vs_client_full_tunnel() { let (_state, iface, peer, _session_id) = setup_test_context().await; // 1. Server-side WireGuard peer AllowedIPs MUST be strictly the assigned client IP (10.100.0.9/32) assert_eq!(peer.server_wireguard_allowed_ips(), "10.100.0.9/32"); // 2. Client configuration MUST contain the FullTunnel routing policy (0.0.0.0/0 for IPv4-only server) let conf = ClientConfigBuilder::build(&peer, &iface, "192.168.1.8:51820").unwrap(); assert!(conf.contains("Address = 10.100.0.9/32")); assert!(conf.contains("AllowedIPs = 0.0.0.0/0")); assert!(conf.contains("Endpoint = 192.168.1.8:51820")); assert!(conf.contains("PersistentKeepalive = 25")); // Dual-stack interface exports dual-stack full tunnel let mut dual_iface = iface.clone(); dual_iface.address_v6 = Some(IpNet::from_str("fd00::1/64").unwrap()); let dual_conf = ClientConfigBuilder::build(&peer, &dual_iface, "192.168.1.8:51820").unwrap(); assert!(dual_conf.contains("AllowedIPs = 0.0.0.0/0, ::/0")); } #[tokio::test] async fn test_endpoint_resolution_failure_and_override() { let (state, _iface, peer, session_id) = setup_test_context().await; let app = build_api_router(state.clone()); // 1. Config export without persistent setting or query endpoint fails with 422 let req = Request::builder() .uri(format!("/api/v1/peers/{}/config", peer.id)) .header("Cookie", format!("nx9_session={session_id}")) .body(Body::empty()) .unwrap(); let res = app.clone().oneshot(req).await.unwrap(); assert_eq!(res.status(), StatusCode::UNPROCESSABLE_ENTITY); let body = axum::body::to_bytes(res.into_body(), usize::MAX) .await .unwrap(); let err_json: serde_json::Value = serde_json::from_slice(&body).unwrap(); assert!( err_json["error"]["message"] .as_str() .unwrap() .contains("No reachable WireGuard server endpoint is configured") ); // 2. Setting persistent server_endpoint setting succeeds state .store .set_setting("server_endpoint", "192.168.1.8:51820", false) .await .unwrap(); let req = Request::builder() .uri(format!("/api/v1/peers/{}/config", peer.id)) .header("Cookie", format!("nx9_session={session_id}")) .body(Body::empty()) .unwrap(); let res = app.clone().oneshot(req).await.unwrap(); assert_eq!(res.status(), StatusCode::OK); let body = axum::body::to_bytes(res.into_body(), usize::MAX) .await .unwrap(); let conf_str = String::from_utf8(body.to_vec()).unwrap(); assert!(conf_str.contains("Endpoint = 192.168.1.8:51820")); // 3. Explicit query parameter overrides persistent setting let req = Request::builder() .uri(format!( "/api/v1/peers/{}/config?server_endpoint=vpn.publicdomain.org:51820", peer.id )) .header("Cookie", format!("nx9_session={session_id}")) .body(Body::empty()) .unwrap(); let res = app.clone().oneshot(req).await.unwrap(); assert_eq!(res.status(), StatusCode::OK); let body = axum::body::to_bytes(res.into_body(), usize::MAX) .await .unwrap(); let conf_str = String::from_utf8(body.to_vec()).unwrap(); assert!(conf_str.contains("Endpoint = vpn.publicdomain.org:51820")); } #[tokio::test] async fn test_learned_endpoint_and_handshake_telemetry_ingestion() { let (state, iface, peer, _session_id) = setup_test_context().await; let wg_engine = Arc::new(SimulatedWireGuardEngine::new()); let net_engine = Arc::new(SimulatedNetworkEngine::new()); // Sync initial state to simulated engine wg_engine .sync_interface(&iface, std::slice::from_ref(&peer)) .await .unwrap(); let reconciler = ReconciliationEngine::new(state.clone(), wg_engine.clone(), net_engine.clone()); // Initially peer has no learned endpoint or handshake in DB let p_db = state.store.get_peer(peer.id).await.unwrap().unwrap(); assert!(p_db.endpoint.is_none()); assert!(p_db.last_handshake_at.is_none()); // Simulate incoming authenticated handshake from client let hs_time = Utc::now().naive_utc(); let learned_client_ep = "192.168.1.50:41234".to_string(); // Apply initial baseline state to ensure all subsystems start converged let initial_report = reconciler.apply().await.unwrap(); assert!(initial_report.success); // Directly simulate kernel stats containing learned endpoint and handshake let live_peers = vec![LivePeerStats { public_key: peer.public_key.as_str().to_string(), endpoint: Some(learned_client_ep.clone()), rx_bytes: 1024, tx_bytes: 2048, last_handshake_at: Some(hs_time), allowed_ips: vec!["10.100.0.9/32".to_string()], persistent_keepalive: Some(25), }]; wg_engine .inject_interface_stats(LiveInterfaceStats { name: iface.name.clone(), public_key: iface.public_key.as_str().to_string(), listen_port: iface.listen_port, fwmark: 0, peers: live_peers, addresses: vec!["10.100.0.1/24".to_string()], mtu: Some(1420), is_up: true, }) .await; // Run reconciliation plan — should ingest telemetry without peer drift let plan = reconciler.plan().await.unwrap(); assert_eq!(plan.peer_changes, 0); // Verify learned telemetry was ingested into the SQLite store let updated_peer = state.store.get_peer(peer.id).await.unwrap().unwrap(); assert_eq!(updated_peer.endpoint.as_deref(), Some("192.168.1.50:41234")); assert_eq!( updated_peer .last_handshake_at .unwrap() .and_utc() .timestamp(), hs_time.and_utc().timestamp() ); } #[tokio::test] async fn test_peer_allowed_ips_and_keepalive_kernel_drift() { let (state, iface, peer, _session_id) = setup_test_context().await; let wg_engine = Arc::new(SimulatedWireGuardEngine::new()); let net_engine = Arc::new(SimulatedNetworkEngine::new()); let reconciler = ReconciliationEngine::new(state.clone(), wg_engine.clone(), net_engine.clone()); // Inject drift: kernel peer erroneously has 0.0.0.0/0 as AllowedIPs and keepalive = 10s let drifted_peers = vec![LivePeerStats { public_key: peer.public_key.as_str().to_string(), endpoint: None, rx_bytes: 0, tx_bytes: 0, last_handshake_at: None, allowed_ips: vec!["0.0.0.0/0".to_string(), "::/0".to_string()], persistent_keepalive: Some(10), }]; wg_engine .inject_interface_stats(LiveInterfaceStats { name: iface.name.clone(), public_key: iface.public_key.as_str().to_string(), listen_port: iface.listen_port, fwmark: 0, peers: drifted_peers, addresses: vec!["10.100.0.1/24".to_string()], mtu: Some(1420), is_up: true, }) .await; // Reconciliation plan MUST detect this semantic drift let plan = reconciler.plan().await.unwrap(); assert!(plan.has_drift); assert_eq!(plan.peer_changes, 1); assert!( plan.actions .iter() .any(|a| a.action_type == "update_peer" && a.description.contains("AllowedIPs drift")) ); // Execute reconciliation apply let report = reconciler.apply().await.unwrap(); assert!(report.success); assert_eq!( report.status, nx9_wg_api::reconciliation::ReconciliationStatus::Converged ); // Verify post-apply convergence: zero drift let post_plan = reconciler.plan().await.unwrap(); assert!(!post_plan.has_drift); assert_eq!(post_plan.peer_changes, 0); // Verify live kernel stats now match desired server AllowedIPs (10.100.0.9/32) let live_stats = wg_engine.get_interface_stats("wg0").await.unwrap().unwrap(); assert_eq!(live_stats.peers[0].allowed_ips, vec!["10.100.0.9/32"]); assert_eq!(live_stats.peers[0].persistent_keepalive, Some(25)); } #[tokio::test] async fn test_forwarding_and_nat_reconciliation_invariants() { let (state, _iface, _peer, _session_id) = setup_test_context().await; let wg_engine = Arc::new(SimulatedWireGuardEngine::new()); let net_engine = Arc::new(SimulatedNetworkEngine::new()); let reconciler = ReconciliationEngine::new(state.clone(), wg_engine.clone(), net_engine.clone()); // Enable NAT masquerade in settings state .store .set_setting("nat_enabled", "true", false) .await .unwrap(); // Reconcile apply let report = reconciler.apply().await.unwrap(); assert!(report.success); assert_eq!( report.status, nx9_wg_api::reconciliation::ReconciliationStatus::Converged ); // Verify NAT masquerade is active in network engine for 10.100.0.0/24 subnet let ruleset = net_engine.get_active_nftables_ruleset().await.unwrap(); assert!(ruleset.contains("masquerade")); assert!(ruleset.contains("10.100.0.0/24")); // Re-planning shows 0 drift let plan = reconciler.plan().await.unwrap(); assert!(!plan.has_drift); assert_eq!(plan.firewall_changes, 0); assert_eq!(plan.route_changes, 0); assert_eq!(plan.interface_changes, 0); } #[tokio::test] async fn test_interface_editing_persistence_and_key_preservation() { let (state, iface, _peer, session_id) = setup_test_context().await; let app = build_api_router(state.clone()); let orig_priv_key = iface.private_key.clone(); let orig_pub_key = iface.public_key.clone(); let orig_id = iface.id; // 1. Edit interface wg0 (change address_v4, listen_port, MTU, DNS, enabled) let update_req = Request::builder() .method("PUT") .uri(format!("/api/v1/interfaces/{}", iface.id)) .header("Content-Type", "application/json") .header("Cookie", format!("nx9_session={session_id}")) .body(Body::from( serde_json::json!({ "name": "wg0", "address_v4": "10.200.0.1/24", "listen_port": 51822, "mtu": 1360, "dns": "9.9.9.9", "enabled": true }) .to_string(), )) .unwrap(); let resp = app.clone().oneshot(update_req).await.unwrap(); assert_eq!(resp.status(), StatusCode::OK); // 2. Query updated interface from database let updated_iface = state.store.get_interface(orig_id).await.unwrap().unwrap(); assert_eq!(updated_iface.address_v4.to_string(), "10.200.0.1/24"); assert_eq!(updated_iface.listen_port, 51822); assert_eq!(updated_iface.mtu, Some(1360)); assert_eq!(updated_iface.dns, Some("9.9.9.9".to_string())); // 3. Verify private key, public key, and ID were strictly preserved (NEVER regenerated) assert_eq!(updated_iface.id, orig_id); assert_eq!(updated_iface.private_key.as_str(), orig_priv_key.as_str()); assert_eq!(updated_iface.public_key.as_str(), orig_pub_key.as_str()); // 4. Verify peers attached to wg0 were preserved let peers = state.store.list_peers_for_interface(orig_id).await.unwrap(); assert_eq!(peers.len(), 1); assert_eq!(peers[0].name, "Mobile"); } #[tokio::test] async fn test_server_endpoint_persistence_validation_and_export_precedence() { let (state, _iface, peer, session_id) = setup_test_context().await; let app = build_api_router(state.clone()); // 1. Invalid server_endpoint format (missing port) is rejected with 422 let invalid_setting_req = Request::builder() .method("PUT") .uri("/api/v1/system/settings") .header("Content-Type", "application/json") .header("Cookie", format!("nx9_session={session_id}")) .body(Body::from( serde_json::json!({ "key": "server_endpoint", "value": "192.168.1.8", // missing port! "is_secret": false }) .to_string(), )) .unwrap(); let resp = app.clone().oneshot(invalid_setting_req).await.unwrap(); assert_eq!(resp.status(), StatusCode::UNPROCESSABLE_ENTITY); // 2. Valid server_endpoint saves successfully let valid_setting_req = Request::builder() .method("PUT") .uri("/api/v1/system/settings") .header("Content-Type", "application/json") .header("Cookie", format!("nx9_session={session_id}")) .body(Body::from( serde_json::json!({ "key": "server_endpoint", "value": "192.168.1.8:51820", "is_secret": false }) .to_string(), )) .unwrap(); let resp = app.clone().oneshot(valid_setting_req).await.unwrap(); assert_eq!(resp.status(), StatusCode::OK); // 3. Export config without override consumes the persisted setting automatically let export_req = Request::builder() .uri(format!("/api/v1/peers/{}/config", peer.id)) .header("Cookie", format!("nx9_session={session_id}")) .body(Body::empty()) .unwrap(); let resp = app.clone().oneshot(export_req).await.unwrap(); assert_eq!(resp.status(), StatusCode::OK); let conf_bytes = axum::body::to_bytes(resp.into_body(), usize::MAX) .await .unwrap(); let conf_str = String::from_utf8(conf_bytes.to_vec()).unwrap(); assert!(conf_str.contains("Endpoint = 192.168.1.8:51820")); // 4. Export QR code returns JSON with SVG and data_url containing the same endpoint let qr_req = Request::builder() .uri(format!("/api/v1/peers/{}/qr", peer.id)) .header("Cookie", format!("nx9_session={session_id}")) .body(Body::empty()) .unwrap(); let resp = app.clone().oneshot(qr_req).await.unwrap(); assert_eq!(resp.status(), StatusCode::OK); let qr_bytes = axum::body::to_bytes(resp.into_body(), usize::MAX) .await .unwrap(); let qr_json: serde_json::Value = serde_json::from_slice(&qr_bytes).unwrap(); assert!(qr_json["svg"].as_str().unwrap().contains(" exports [2001:db8::10]:51820 let set_v6_req = Request::builder() .method("PUT") .uri("/api/v1/system/settings") .header("Content-Type", "application/json") .header("Cookie", format!("nx9_session={session_id}")) .body(Body::from( serde_json::json!({ "key": "wireguard.server_host", "value": "2001:db8::10", "is_secret": false }) .to_string(), )) .unwrap(); let resp = app.clone().oneshot(set_v6_req).await.unwrap(); assert_eq!(resp.status(), StatusCode::OK); let export_v6_req = Request::builder() .uri(format!("/api/v1/peers/{}/config", peer.id)) .header("Cookie", format!("nx9_session={session_id}")) .body(Body::empty()) .unwrap(); let resp = app.clone().oneshot(export_v6_req).await.unwrap(); assert_eq!(resp.status(), StatusCode::OK); let conf_bytes = axum::body::to_bytes(resp.into_body(), usize::MAX) .await .unwrap(); let conf_str = String::from_utf8(conf_bytes.to_vec()).unwrap(); assert!(conf_str.contains("Endpoint = [2001:db8::10]:51820")); } #[tokio::test] async fn test_peer_telemetry_enrichment_and_status_transitions() { let (state_orig, iface, peer, session_id) = setup_test_context().await; let simulated_wg = Arc::new(SimulatedWireGuardEngine::new()); let simulated_net = Arc::new(SimulatedNetworkEngine::new()); let state = AppState::with_engines( state_orig.store.clone(), simulated_wg.clone(), simulated_net.clone(), ); // Inject live kernel statistics into simulated WireGuard engine let recent_hs = Utc::now().naive_utc() - chrono::Duration::seconds(15); let live_peer = LivePeerStats { public_key: peer.public_key.to_string(), endpoint: Some("192.168.1.50:41234".to_string()), rx_bytes: 409600, tx_bytes: 819200, last_handshake_at: Some(recent_hs), allowed_ips: vec!["10.100.0.9/32".to_string()], persistent_keepalive: Some(25), }; let live_iface = LiveInterfaceStats { name: iface.name.clone(), public_key: iface.public_key.to_string(), listen_port: iface.listen_port, fwmark: 0, peers: vec![live_peer], addresses: vec!["10.100.0.1/24".to_string()], mtu: Some(1420), is_up: true, }; // Inject live stats into simulated_wg simulated_wg.inject_interface_stats(live_iface).await; let app = build_api_router(state.clone()); // Query GET /api/v1/peers let list_req = Request::builder() .uri("/api/v1/peers") .header("Cookie", format!("nx9_session={session_id}")) .body(Body::empty()) .unwrap(); let resp = app.clone().oneshot(list_req).await.unwrap(); assert_eq!(resp.status(), StatusCode::OK); let body_bytes = axum::body::to_bytes(resp.into_body(), usize::MAX) .await .unwrap(); let peers_json: Vec = serde_json::from_slice(&body_bytes).unwrap(); assert_eq!(peers_json.len(), 1); let p = &peers_json[0]; assert_eq!(p["name"], "Mobile"); assert_eq!(p["endpoint"], "192.168.1.50:41234"); assert_eq!(p["rx_bytes"], 409600); assert_eq!(p["tx_bytes"], 819200); // Handshake is serialized as explicit RFC3339 UTC string with offset/Z let hs_str = p["last_handshake_at"].as_str().unwrap(); assert!(hs_str.contains('T')); assert!(hs_str.ends_with('Z') || hs_str.contains("+00:00")); // Verify learned telemetry was cached in SQLite let db_peer = state.store.get_peer(peer.id).await.unwrap().unwrap(); assert_eq!(db_peer.endpoint, Some("192.168.1.50:41234".to_string())); assert_eq!( db_peer.last_handshake_at.unwrap().and_utc().timestamp(), recent_hs.and_utc().timestamp() ); }