Coverage for integrations / remote_desktop / device_id.py: 71.4%

1""" 

2Device Identity — Deterministic device IDs for remote desktop sessions. 

3 

4Reuses compute_mesh_service.py:116 pattern: device_id = SHA256(public_key)[:16]. 

5Device IDs are user-scoped (tied to user_id), displayed in 3-group format (847-291-053). 

6""" 

7 

8import hashlib 

9import logging 

10import os 

11import platform 

12import uuid 

13from typing import Optional 

14 

15logger = logging.getLogger('hevolve.remote_desktop') 

16 

17# ── Device ID cache ───────────────────────────────────────────── 

18_cached_device_id: Optional[str] = None 

19 

20 

21def _resolve_key_dir() -> str: 

22 """Resolve key directory — same logic as compute_mesh_service.py.""" 

23 data_dir = os.environ.get('HEVOLVE_DATA_DIR', '') 

24 if data_dir: 

25 return os.path.join(data_dir, 'mesh', 'keys') 

26 # Fallback: agent_data in project root or user home 

27 # Try platform_paths first for cross-platform support 

28 try: 

29 from core.platform_paths import get_agent_data_dir 

30 _pp_agent_dir = get_agent_data_dir() 

31 except ImportError: 

32 _pp_agent_dir = os.path.join(os.path.expanduser('~'), 'Documents', 'Nunba', 'data', 'agent_data') 

33 for candidate in [ 

34 os.path.join(os.path.dirname(os.path.dirname(os.path.dirname( 

35 os.path.abspath(__file__)))), 'agent_data'), 

36 _pp_agent_dir, 

37 ]: 

38 if os.path.isdir(candidate): 

39 return candidate 

40 return os.path.join(os.path.expanduser('~'), '.hart', 'keys') 

41 

42 

43def _generate_machine_fingerprint() -> str: 

44 """Generate stable machine fingerprint when no public key file exists. 

45 

46 Uses platform + hostname + MAC address — deterministic per machine. 

47 """ 

48 components = [ 

49 platform.node(), 

50 platform.machine(), 

51 platform.system(), 

52 str(uuid.getnode()), # MAC address as int 

53 ] 

54 return '|'.join(components) 

55 

56 

57def get_device_id() -> str: 

58 """Get this device's 16-char hex ID. 

59 

60 Priority: 

61 1. Public key file (compute_mesh_service.py:116 pattern) 

62 2. Machine fingerprint fallback (deterministic) 

63 

64 Returns: 

65 16-character hex string (e.g., '847291053def3a21') 

66 """ 

67 global _cached_device_id 

68 if _cached_device_id is not None: 

69 return _cached_device_id 

70 

71 key_dir = _resolve_key_dir() 

72 

73 # Try public key file first (same as compute_mesh_service.py:113-116) 

74 for key_filename in ('public.key', 'node_public.key', 'node_x25519_public.key'): 

75 key_path = os.path.join(key_dir, key_filename) 

76 if os.path.exists(key_path): 

77 try: 

78 with open(key_path, 'r') as f: 

79 pub_key = f.read().strip() 

80 if pub_key: 

81 _cached_device_id = hashlib.sha256(pub_key.encode()).hexdigest()[:16] 

82 logger.info(f"Device ID from {key_filename}: {_cached_device_id}") 

83 return _cached_device_id 

84 except (OSError, UnicodeDecodeError): 

85 continue 

86 

87 # Fallback: machine fingerprint (deterministic per machine) 

88 fingerprint = _generate_machine_fingerprint() 

89 _cached_device_id = hashlib.sha256(fingerprint.encode()).hexdigest()[:16] 

90 logger.info(f"Device ID from fingerprint: {_cached_device_id}") 

91 return _cached_device_id 

92 

93 

94def format_device_id(device_id: str) -> str: 

95 """Format 16-char hex ID for display: '847291053def3a21' → '847-291-053'. 

96 

97 Uses first 9 hex chars split into groups of 3 (like AnyDesk's numeric IDs). 

98 """ 

99 digits = device_id[:9] 

100 return f"{digits[:3]}-{digits[3:6]}-{digits[6:9]}" 

101 

102 

103def parse_device_id(formatted: str) -> str: 

104 """Parse display-formatted ID back to lookup key. 

105 

106 '847-291-053' → '847291053' (prefix match against full 16-char IDs). 

107 """ 

108 return formatted.replace('-', '').replace(' ', '').lower() 

109 

110 

111def get_user_device_id(user_id: str) -> str: 

112 """Get device ID scoped to a user (for cross-device lookup). 

113 

114 Combines machine device_id with user_id for user-scoped identity. 

115 This matches the proximity_service.py:41 pattern where device_id 

116 is used for cross-device dedup per user. 

117 """ 

118 raw_device = get_device_id() 

119 return hashlib.sha256(f"{user_id}:{raw_device}".encode()).hexdigest()[:16] 

120 

121 

122def register_device(user_id: str, device_id: Optional[str] = None) -> dict: 

123 """Register this device for a user (enables cross-device discovery). 

124 

125 Reuses PeerNode model from integrations/social/models.py:580 

126 (node_operator_id FK→User). 

127 

128 Returns: 

129 {'device_id': str, 'user_id': str, 'registered': bool} 

130 """ 

131 dev_id = device_id or get_device_id() 

132 try: 

133 from integrations.social.models import db_session, PeerNode 

134 with db_session() as db: 

135 existing = db.query(PeerNode).filter( 

136 PeerNode.node_id == dev_id, 

137 ).first() 

138 if existing: 

139 existing.node_operator_id = int(user_id) if user_id.isdigit() else None 

140 existing.status = 'active' 

141 else: 

142 node = PeerNode( 

143 node_id=dev_id, 

144 url=f'localhost:{os.environ.get("HART_PORT", "6777")}', 

145 node_operator_id=int(user_id) if user_id.isdigit() else None, 

146 status='active', 

147 ) 

148 db.add(node) 

149 db.commit() 

150 logger.info(f"Device {dev_id} registered for user {user_id}") 

151 return {'device_id': dev_id, 'user_id': user_id, 'registered': True} 

152 except Exception as e: 

153 logger.warning(f"Device registration failed (DB unavailable): {e}") 

154 return {'device_id': dev_id, 'user_id': user_id, 'registered': False} 

155 

156 

157def discover_user_devices(user_id: str) -> list: 

158 """Find all devices registered to a user. 

159 

160 Queries PeerNode.node_operator_id (models.py:580) — same as 

161 compute_mesh_service.py:131 discover_peers(). 

162 

163 Returns: 

164 List of {'device_id': str, 'url': str, 'status': str, 'last_seen': str} 

165 """ 

166 try: 

167 from integrations.social.models import db_session, PeerNode 

168 with db_session() as db: 

169 nodes = db.query(PeerNode).filter( 

170 PeerNode.node_operator_id == (int(user_id) if user_id.isdigit() else -1), 

171 PeerNode.status == 'active', 

172 ).all() 

173 return [ 

174 { 

175 'device_id': n.node_id, 

176 'url': n.url, 

177 'status': n.status, 

178 'last_seen': str(n.last_seen) if n.last_seen else None, 

179 } 

180 for n in nodes 

181 ] 

182 except Exception as e: 

183 logger.warning(f"Device discovery failed (DB unavailable): {e}") 

184 return []