Coverage for security/native_hive

1"""

2Native Hive AI Loader — Loads closed-source HevolveAI at runtime.

4Architecture:

5 HART OS (open source, BSL-1.1)

6 │

7 ├─ Reads source code freely — it's open source

8 │

9 └─ Loads HevolveAI (closed source, compiled)

10 │

11 ├─ PRIMARY: Cython-compiled Python wheel

12 │ ├─ .so (Linux) / .pyd (Windows) — standard Python C extensions

13 │ ├─ Compiled from Python via Cython — not readable source

14 │ ├─ Installed: pip install hevolveai-0.1.0-cp311-cp311-linux_x86_64.whl

15 │ └─ Imported via: import hevolveai (standard Python import)

16 │

17 ├─ FALLBACK: Native binary via ctypes (Rust hevolveai_topo)

18 │ ├─ .so (Linux) / .dll (Windows) / .dylib (macOS)

19 │ ├─ Signed by master key — tampered binaries rejected

20 │ └─ Exposed via C ABI + Python ctypes wrapper

21 │

22 ├─ Provides: Hebbian learning, Bayesian inference,

23 │ RALT distribution, world model, embodied AI

24 │

25 └─ STUB: Pure-Python stubs if neither path available

27Who can run HevolveAI:

28 Anyone on a legitimate deployment — Nunba, HARTOS standalone, Docker,

29 HART OS (Live OS), cloud, or pip install. Flat, regional, or central tier.

30 Users never need the master key. They download already-signed binaries.

32Protection (against forks weaponizing, NOT against users running):

33 1. Cython-compiled .so/.pyd — not readable Python, not trivially decompilable

34 2. Release-signed by master key — proves authenticity (not tampered)

35 3. Origin attestation check — refuses to load on unauthorized forks

36 4. Forks cannot sign modified packages — no master key = can't distribute

37 5. Forks cannot join federation — origin attestation fails

38 6. License prohibits decompilation / reverse engineering

40Load order:

41 1. import hevolveai (Cython-compiled wheel, pip-installed)

42 2. HEVOLVE_NATIVE_LIB env var (ctypes binary, explicit path)

43 3. /usr/lib/hevolve/libhevolve_ai.so (system install)

44 4. ~/.hevolve/lib/libhevolve_ai.so (user install)

45 5. {HART_ROOT}/lib/libhevolve_ai.so (in-tree)

46 6. Falls back to pure-Python stubs (reduced functionality)

47"""

49import ctypes

50import hashlib

51import json

52import logging

53import os

54import platform

55import struct

56import sys

57from pathlib import Path

58from typing import Any, Callable, Dict, Optional, Tuple

60logger = logging.getLogger('hevolve_security')

62# ═══════════════════════════════════════════════════════════════════════

63# Binary naming convention per platform

64# ═══════════════════════════════════════════════════════════════════════

66_PLATFORM_LIB = {

67 'Linux': 'libhevolve_ai.so',

68 'Darwin': 'libhevolve_ai.dylib',

69 'Windows': 'hevolve_ai.dll',

70}

72_LIB_NAME = _PLATFORM_LIB.get(platform.system(), 'libhevolve_ai.so')

74_HART_ROOT = os.environ.get(

75 'HART_INSTALL_DIR',

76 os.path.dirname(os.path.dirname(os.path.abspath(__file__)))

77)

79# Search paths for the native binary

80_SEARCH_PATHS = [

81 os.environ.get('HEVOLVE_NATIVE_LIB', ''),

82 f'/usr/lib/hevolve/{_LIB_NAME}',

83 f'/usr/local/lib/hevolve/{_LIB_NAME}',

84 os.path.expanduser(f'~/.hevolve/lib/{_LIB_NAME}'),

85 os.path.join(_HART_ROOT, 'lib', _LIB_NAME),

86]

88# Expected signatures for known binary versions (SHA-256 of binary)

89# Updated on each official release by the CI/CD pipeline

90_KNOWN_SIGNATURES: Dict[str, str] = {}

92# Module-level singleton

93_native_lib: Optional[ctypes.CDLL] = None

94_cython_module = None # The Cython-compiled hevolveai package (if imported)

95_native_available = False

96_stub_mode = False

97_load_method: Optional[str] = None # 'cython' | 'ctypes' | None

100# ═══════════════════════════════════════════════════════════════════════

101# Binary verification

102# ═══════════════════════════════════════════════════════════════════════

103

104def _compute_binary_hash(path: str) -> str:

105 """SHA-256 hash of the native binary file."""

106 h = hashlib.sha256()

107 with open(path, 'rb') as f:

108 for chunk in iter(lambda: f.read(65536), b''):

109 h.update(chunk)

110 return h.hexdigest()

111

112

113def _verify_binary_signature(path: str) -> Tuple[bool, str]:

114 """Verify the native binary was signed by the master key.

115

116 The binary has a 256-byte Ed25519 signature appended after a magic marker.

117 Format: [binary content] [HEVOLVE_SIG_V1] [64-byte Ed25519 signature]

118 """

119 MAGIC = b'HEVOLVE_SIG_V1'

120 SIG_LEN = 64 # Ed25519 signature is 64 bytes

121

122 try:

123 with open(path, 'rb') as f:

124 f.seek(0, 2)

125 file_size = f.tell()

126

127 if file_size < len(MAGIC) + SIG_LEN + 1024:

128 return False, 'Binary too small to contain signature'

129

130 # Read the signature trailer

131 trailer_size = len(MAGIC) + SIG_LEN

132 f.seek(-trailer_size, 2)

133 trailer = f.read(trailer_size)

134

135 if not trailer.startswith(MAGIC):

136 # No signature embedded — check known hashes instead

137 bin_hash = _compute_binary_hash(path)

138 if bin_hash in _KNOWN_SIGNATURES:

139 return True, f'Known binary hash: {bin_hash[:16]}...'

140 # In dev mode, allow unsigned binaries

141 if os.environ.get('HEVOLVE_DEV_MODE', '').lower() == 'true':

142 return True, 'Dev mode — unsigned binary allowed'

143 return False, 'No embedded signature and unknown hash'

144

145 # Extract signature

146 sig_bytes = trailer[len(MAGIC):]

147

148 # Hash the binary content (excluding the signature trailer)

149 content_size = file_size - trailer_size

150 f.seek(0)

151 h = hashlib.sha256()

152 remaining = content_size

153 while remaining > 0:

154 chunk = f.read(min(65536, remaining))

155 if not chunk:

156 break

157 h.update(chunk)

158 remaining -= len(chunk)

159 content_hash = h.digest()

160

161 # Verify with master public key

162 from security.master_key import get_master_public_key

163 pub_key = get_master_public_key()

164 pub_key.verify(sig_bytes, content_hash)

165 return True, 'Binary signature verified by master key'

166

167 except Exception as e:

168 return False, f'Signature verification error: {e}'

169

170

171def _verify_binary_origin_check(lib: ctypes.CDLL) -> Tuple[bool, str]:

172 """Call the binary's built-in origin check function.

173

174 The native binary has a compiled-in function that verifies

175 it's running inside genuine HART OS (checks origin fingerprint).

176 A fork loading the binary will fail this check.

177 """

178 try:

179 # The binary exposes: int hevolve_verify_origin(const char* fingerprint)

180 # Returns 0 on success, non-zero on failure

181 func = lib.hevolve_verify_origin

182 func.argtypes = [ctypes.c_char_p]

183 func.restype = ctypes.c_int

184

185 from security.origin_attestation import ORIGIN_FINGERPRINT

186 result = func(ORIGIN_FINGERPRINT.encode('utf-8'))

187 if result == 0:

188 return True, 'Binary origin check passed'

189 return False, f'Binary rejected origin (code {result})'

190 except AttributeError:

191 # Function not exported — older binary version, accept

192 return True, 'Binary does not export origin check (older version)'

193 except Exception as e:

194 return False, f'Origin check error: {e}'

195

196

197# ═══════════════════════════════════════════════════════════════════════

198# Binary loading

199# ═══════════════════════════════════════════════════════════════════════

200

201def _find_native_binary() -> Optional[str]:

202 """Search for the native binary in known locations.

203

204 Checks for both plaintext (.so/.dll) and encrypted (.so.enc) variants.

205 Encrypted binaries are decrypted to tmpfs (RAM) at load time.

206 """

207 for path in _SEARCH_PATHS:

208 if path and os.path.isfile(path):

209 return path

210 # Check for encrypted variant

211 enc_path = path + '.enc' if path else ''

212 if enc_path and os.path.isfile(enc_path):

213 decrypted = _decrypt_binary_to_tmpfs(enc_path)

214 if decrypted:

215 return decrypted

216 return None

217

218

219def _decrypt_binary_to_tmpfs(enc_path: str) -> Optional[str]:

220 """Decrypt an encrypted HevolveAI binary to RAM-only filesystem.

221

222 The binary is AES-256-GCM encrypted. The decryption key is derived via

223 ECDH between this node's Ed25519 key and the master public key, ensuring

224 only legitimate HART OS nodes (with valid first-boot keypairs) can decrypt.

225

226 File format: [12-byte nonce][ciphertext][16-byte GCM tag]

227

228 The decrypted binary lives ONLY in tmpfs (RAM) — never touches disk.

229 Docker: --read-only --tmpfs /run/hevolve ensures this.

230 """

231 try:

232 from cryptography.hazmat.primitives.ciphers.aead import AESGCM

233 except ImportError:

234 logger.warning("cryptography package not available — cannot decrypt binary")

235 return None

236

237 try:

238 # Derive decryption key: HKDF(node_private_key_seed || master_public_key)

239 # This ensures only nodes with valid Ed25519 keys from first-boot can decrypt

240 node_key_path = os.path.expanduser('~/.hevolve/keys/node_private.key')

241 if not os.path.isfile(node_key_path):

242 node_key_path = '/var/lib/hevolve/node_private.key'

243 if not os.path.isfile(node_key_path):

244 logger.debug("No node private key — cannot decrypt binary")

245 return None

246

247 with open(node_key_path, 'rb') as f:

248 node_seed = f.read(32)

249

250 from security.master_key import MASTER_PUBLIC_KEY_HEX

251 master_pub_bytes = bytes.fromhex(MASTER_PUBLIC_KEY_HEX)

252

253 # HKDF: derive AES-256 key from node_seed + master_public_key

254 from cryptography.hazmat.primitives.kdf.hkdf import HKDF

255 from cryptography.hazmat.primitives import hashes

256 hkdf = HKDF(

257 algorithm=hashes.SHA256(),

258 length=32,

259 salt=master_pub_bytes,

260 info=b'hevolve-native-binary-v1',

261 )

262 aes_key = hkdf.derive(node_seed)

263

264 # Read encrypted binary

265 with open(enc_path, 'rb') as f:

266 data = f.read()

267

268 if len(data) < 28: # 12 nonce + 16 tag minimum

269 logger.warning(f"Encrypted binary too small: {enc_path}")

270 return None

271

272 nonce = data[:12]

273 ciphertext_with_tag = data[12:]

274

275 # Decrypt

276 aesgcm = AESGCM(aes_key)

277 plaintext = aesgcm.decrypt(nonce, ciphertext_with_tag, None)

278

279 # Write to tmpfs (RAM only — never persists to disk)

280 tmpfs_dir = os.environ.get('HEVOLVE_TMPFS', '/run/hevolve')

281 if not os.path.isdir(tmpfs_dir):

282 # Fallback: /dev/shm (Linux shared memory) or system temp

283 for candidate in ['/dev/shm/hevolve', '/tmp/.hevolve_runtime']:

284 try:

285 os.makedirs(candidate, mode=0o700, exist_ok=True)

286 tmpfs_dir = candidate

287 break

288 except OSError:

289 continue

290

291 decrypted_path = os.path.join(tmpfs_dir, _LIB_NAME)

292 with open(decrypted_path, 'wb') as f:

293 f.write(plaintext)

294 os.chmod(decrypted_path, 0o500) # read+execute only

295

296 logger.info(f"Decrypted HevolveAI binary to tmpfs: {decrypted_path}")

297 return decrypted_path

298

299 except Exception as e:

300 logger.warning(f"Failed to decrypt binary {enc_path}: {e}")

301 return None

302

303

304_armor_install_tried = False

305_armor_install_ok = False

306

307

308def _try_install_hevolvearmor() -> Tuple[bool, str]:

309 """Install the HevolveArmor import hook if an armored bundle is present.

310

311 Looks for a HevolveArmor-armored hevolveai at one of:

312 - $HEVOLVE_ARMORED_DIR (explicit override)

313 - {HART_ROOT}/vendor/hevolveai_armored/modules

314 - {HART_ROOT}/hevolveai_armored/modules

315 - ~/.hevolveai/armored/modules

316 - /opt/hevolveai/armored/modules

317

318 Key resolution (in order):

319 1. $HEVOLVE_ARMOR_KEY_FILE — path to raw 32-byte or hex-encoded key file

320 2. $HEVOLVE_ARMOR_PASSPHRASE — PBKDF2-derived from passphrase + salt file

321 3. derive_runtime_key() — HevolveArmor's machine-bound derivation

322

323 Idempotent: safe to call repeatedly; returns cached result after first try.

324

325 Returns (installed, message).

326 """

327 global _armor_install_tried, _armor_install_ok

328 if _armor_install_tried:

329 return _armor_install_ok, 'armor install cached'

330 _armor_install_tried = True

331

332 try:

333 import hevolvearmor

334 except ImportError:

335 _armor_install_ok = False

336 return False, 'hevolvearmor package not installed'

337

338 # Locate encrypted modules directory

339 candidate_dirs = [

340 os.environ.get('HEVOLVE_ARMORED_DIR', ''),

341 os.path.join(_HART_ROOT, 'vendor', 'hevolveai_armored', 'modules'),

342 os.path.join(_HART_ROOT, 'hevolveai_armored', 'modules'),

343 os.path.expanduser('~/.hevolveai/armored/modules'),

344 '/opt/hevolveai/armored/modules',

345 ]

346 modules_dir = None

347 for d in candidate_dirs:

348 if d and os.path.isdir(d):

349 # Must contain at least one .enc file under hevolveai/

350 hevolveai_dir = os.path.join(d, 'hevolveai')

351 if os.path.isdir(hevolveai_dir):

352 modules_dir = d

353 break

354 if modules_dir is None:

355 _armor_install_ok = False

356 return False, 'no armored hevolveai bundle found'

357

358 # Resolve key

359 key: Optional[bytes] = None

360 key_file = os.environ.get('HEVOLVE_ARMOR_KEY_FILE', '')

361 if key_file and os.path.isfile(key_file):

362 try:

363 with open(key_file, 'rb') as f:

364 raw = f.read().strip()

365 if len(raw) == 32:

366 key = raw

367 elif len(raw) == 64:

368 key = bytes.fromhex(raw.decode('ascii'))

369 except Exception as e:

370 logger.debug(f"[armor] key file read failed: {e}")

371

372 if key is None:

373 passphrase = os.environ.get('HEVOLVE_ARMOR_PASSPHRASE', '')

374 if passphrase:

375 try:

376 key = hevolvearmor.armor_derive_key_passphrase(

377 passphrase, b'hevolveai-armor-salt-v1')

378 except Exception as e:

379 logger.debug(f"[armor] passphrase derivation failed: {e}")

380

381 if key is None:

382 try:

383 key = hevolvearmor.derive_runtime_key()

384 except Exception as e:

385 logger.debug(f"[armor] runtime key derivation failed: {e}")

386

387 if key is None or len(bytes(key)) != 32:

388 _armor_install_ok = False

389 return False, 'no usable armor decryption key'

390

391 try:

392 hevolvearmor.install(modules_dir, bytes(key))

393 _armor_install_ok = True

394 logger.info(

395 f"[armor] HevolveArmor import hook installed: modules_dir={modules_dir}")

396 return True, f'armor installed: {modules_dir}'

397 except Exception as e:

398 _armor_install_ok = False

399 return False, f'armor install failed: {e}'

400

401

402def _try_load_cython_package() -> Tuple[bool, str]:

403 """Try to import the Cython-compiled hevolveai wheel.

404

405 This is the PRIMARY load path. The wheel is installed via pip and

406 contains .so/.pyd Cython extensions — standard Python imports, no ctypes.

407

408 Before attempting the import, we call `_try_install_hevolvearmor()` so

409 that an armored bundle (if present) is transparently decryptable via

410 the import hook.

411

412 Returns (success, message).

413 """

414 global _cython_module, _native_available, _stub_mode, _load_method

415

416 # Install armor import hook first (no-op if already tried or absent).

417 _try_install_hevolvearmor()

418

419 try:

420 import hevolveai

421 # Verify it's actually compiled (not someone's source checkout)

422 mod_file = getattr(hevolveai, '__file__', '') or ''

423 # Compiled packages have __init__.cpython-*.so or __init__.pyd

424 # OR a minimal stub __init__.py that imports from compiled submodules

425 # Check for at least one compiled submodule

426 pkg_dir = os.path.dirname(mod_file)

427 if pkg_dir:

428 has_compiled = any(

429 f.endswith('.so') or f.endswith('.pyd')

430 for d, _, files in os.walk(pkg_dir)

431 for f in files

432 if '.cpython-' in f or f.endswith('.pyd')

433 )

434 if not has_compiled:

435 # This is a source checkout, not the compiled wheel

436 return False, 'hevolveai found but not Cython-compiled (source install)'

437

438 _cython_module = hevolveai

439 _native_available = True

440 _stub_mode = False

441 _load_method = 'cython'

442 version = getattr(hevolveai, '__version__', 'unknown')

443 logger.info(f"Loaded HevolveAI Cython package v{version} from {mod_file}")

444 return True, f'Cython wheel loaded: {mod_file}'

445

446 except ImportError:

447 return False, 'hevolveai package not installed'

448 except Exception as e:

449 return False, f'hevolveai import error: {e}'

450

451

452def load_native_lib(force_reload: bool = False) -> Tuple[bool, str]:

453 """Load HevolveAI — tries Cython wheel first, then ctypes binary.

454

455 Returns (success, message).

456

457 Load order:

458 1. Cython-compiled wheel (pip install hevolveai-*.whl)

459 2. ctypes native binary (.so/.dll)

460 3. Stub mode (reduced functionality)

461 """

462 global _native_lib, _native_available, _stub_mode, _load_method

463

464 if (_native_available or _cython_module) and not force_reload:

465 return True, f'Already loaded ({_load_method})'

466

467 # ── Path 1: Cython-compiled Python wheel (primary) ────────

468 ok, msg = _try_load_cython_package()

469 if ok:

470 return True, msg

471 logger.debug(f"Cython path: {msg}")

472

473 # ── Path 2: ctypes native binary (fallback) ──────────────

474 path = _find_native_binary()

475 if not path:

476 _stub_mode = True

477 _native_available = False

478 _load_method = None

479 logger.info(

480 "HevolveAI not available — running in stub mode. "

481 "AI features (Hebbian learning, Bayesian inference, RALT, "

482 "world model) will use pure-Python fallbacks with reduced "

483 "performance. Install the compiled wheel for full capability: "

484 "pip install hevolveai-*.whl"

485 )

486 return False, 'HevolveAI not found — stub mode active'

487

488 # Verify binary signature

489 sig_ok, sig_msg = _verify_binary_signature(path)

490 enforcement = os.environ.get('HEVOLVE_ENFORCEMENT_MODE', 'warn').lower()

491 if not sig_ok:

492 if enforcement == 'hard':

493 _stub_mode = True

494 _native_available = False

495 _load_method = None

496 logger.critical(f"Refusing to load unsigned binary: {sig_msg}")

497 return False, f'Binary signature verification failed: {sig_msg}'

498 else:

499 logger.warning(f"Binary signature warning: {sig_msg}")

500

501 # Load the binary

502 try:

503 _native_lib = ctypes.CDLL(path)

504 logger.info(f"Loaded HevolveAI native binary from {path}")

505 except OSError as e:

506 _stub_mode = True

507 _native_available = False

508 _load_method = None

509 logger.error(f"Failed to load native binary: {e}")

510 return False, f'Load failed: {e}'

511

512 # Origin check — the binary verifies it's inside genuine HART OS

513 origin_ok, origin_msg = _verify_binary_origin_check(_native_lib)

514 if not origin_ok:

515 logger.warning(f"Binary origin check: {origin_msg}")

516 if enforcement == 'hard':

517 _native_lib = None

518 _stub_mode = True

519 _native_available = False

520 _load_method = None

521 return False, f'Binary origin check failed: {origin_msg}'

522

523 # Initialize

524 try:

525 init_func = _native_lib.hevolve_init

526 init_func.restype = ctypes.c_int

527 result = init_func()

528 if result != 0:

529 logger.warning(f"hevolve_init() returned {result}")

530 except AttributeError:

531 pass # No init function — older binary

532

533 _native_available = True

534 _stub_mode = False

535 _load_method = 'ctypes'

536 return True, f'Loaded ctypes binary from {path}'

537

538

539def is_native_available() -> bool:

540 """Check if the native binary is loaded and functional."""

541 return _native_available

542

543

544def is_stub_mode() -> bool:

545 """Check if running with pure-Python stubs (no native binary)."""

546 return _stub_mode

547

548

549def get_native_lib() -> Optional[ctypes.CDLL]:

550 """Get the loaded native library handle."""

551 return _native_lib

552

553

554# ═══════════════════════════════════════════════════════════════════════

555# Python-side function wrappers (safe to call even in stub mode)

556# ═══════════════════════════════════════════════════════════════════════

557

558def get_hevolveai():

559 """Get the loaded hevolveai Cython package, or None.

560

561 HARTOS code that needs HevolveAI should use this:

562 hevolveai = get_hevolveai()

563 if hevolveai:

564 from hevolveai.embodied_ai.learning.hive_mind import HiveMind

565 ...

566 """

567 return _cython_module

568

569

570def native_infer(prompt: str, model: str = 'default',

571 options: Optional[Dict] = None) -> Optional[str]:

572 """Call inference if available, else return None for Python fallback."""

573 # Cython path — call the Python API directly

574 if _cython_module:

575 try:

576 from hevolveai.embodied_ai.inference.qwen_inference_only import infer

577 return infer(prompt, model=model, **(options or {}))

578 except (ImportError, AttributeError):

579 pass

580 except Exception as e:

581 logger.debug(f"Cython inference error: {e}")

582 return None

583

584 # ctypes path

585 if not _native_available or not _native_lib:

586 return None

587 try:

588 func = _native_lib.hevolve_infer

589 func.argtypes = [ctypes.c_char_p, ctypes.c_char_p]

590 func.restype = ctypes.c_char_p

591 result = func(prompt.encode('utf-8'), model.encode('utf-8'))

592 return result.decode('utf-8') if result else None

593 except Exception as e:

594 logger.debug(f"Native inference error: {e}")

595 return None

596

597

598def native_hebbian_update(signals: Dict[str, float]) -> Optional[Dict]:

599 """Run Hebbian learning step."""

600 # Cython path

601 if _cython_module:

602 try:

603 from hevolveai.embodied_ai.learning.hebbian_differentiator import HebbianDifferentiator

604 heb = HebbianDifferentiator()

605 return heb.update(signals)

606 except (ImportError, AttributeError):

607 pass

608 except Exception as e:

609 logger.debug(f"Cython Hebbian error: {e}")

610 return None

611

612 # ctypes path

613 if not _native_available or not _native_lib:

614 return None

615 try:

616 func = _native_lib.hevolve_hebbian_update

617 func.argtypes = [ctypes.c_char_p]

618 func.restype = ctypes.c_char_p

619 payload = json.dumps(signals).encode('utf-8')

620 result = func(payload)

621 return json.loads(result.decode('utf-8')) if result else None

622 except Exception as e:

623 logger.debug(f"Native Hebbian error: {e}")

624 return None

625

626

627def native_version() -> Optional[str]:

628 """Get HevolveAI version string."""

629 if _cython_module:

630 return getattr(_cython_module, '__version__', '0.1.0')

631 if not _native_available or not _native_lib:

632 return None

633 try:

634 func = _native_lib.hevolve_version

635 func.restype = ctypes.c_char_p

636 result = func()

637 return result.decode('utf-8') if result else None

638 except Exception:

639 return None

640

641

642def shutdown_native():

643 """Clean shutdown of HevolveAI."""

644 global _native_lib, _native_available, _cython_module, _load_method

645 if _native_lib:

646 try:

647 func = _native_lib.hevolve_shutdown

648 func.restype = ctypes.c_int

649 func()

650 except AttributeError:

651 pass

652 _native_lib = None

653 _cython_module = None

654 _native_available = False

655 _load_method = None

656

657

658def get_status() -> Dict:

659 """Status summary for diagnostics."""

660 return {

661 'native_available': _native_available,

662 'stub_mode': _stub_mode,

663 'load_method': _load_method,

664 'binary_path': _find_native_binary(),

665 'cython_package': bool(_cython_module),

666 'armor_installed': _armor_install_ok,

667 'version': native_version(),

668 'platform_lib': _LIB_NAME,

669 'search_paths': [p for p in _SEARCH_PATHS if p],

670 }

671

672

673# ═══════════════════════════════════════════════════════════════════════

674# Canonical HevolveAI submodule loader (G2 — unified probe)

675# ═══════════════════════════════════════════════════════════════════════

676

677def try_import_hevolveai(dotted_path: str = 'hevolveai') -> Optional[Any]:

678 """Canonical helper to import HevolveAI (or a submodule) with armor support.

679

680 Replaces scattered ``try: import hevolveai.X except ImportError`` probes

681 across HARTOS (vision/frame_store, agent_engine/world_model_bridge,

682 hart_intelligence_entry, etc.) with a single entry point that:

683

684 1. Ensures HevolveArmor's import hook is installed (if an armored

685 bundle is present on disk and decryption keys are resolvable).

686 2. Guarantees the top-level ``hevolveai`` package is loaded exactly

687 once via the same code path as ``load_native_lib()``.

688 3. Imports the requested dotted submodule via ``importlib``.

689

690 Returns the imported module object, or ``None`` on any failure.

691 All failures are logged at debug level — callers decide whether to

692 log at a higher level based on their fallback semantics.

693

694 Examples:

695 >>> visual_encoding = try_import_hevolveai(

696 ... 'hevolveai.embodied_ai.utils.visual_encoding')

697 >>> if visual_encoding is None:

698 ... # fall back to local numpy implementation

699 ... ...

700

701 >>> hive_mind = try_import_hevolveai(

702 ... 'hevolveai.embodied_ai.learning.hive_mind')

703 """

704 if not dotted_path or not dotted_path.startswith('hevolveai'):

705 logger.debug(f"[try_import_hevolveai] invalid path: {dotted_path!r}")

706 return None

707

708 # Ensure the top-level package is loaded via the canonical path so

709 # armor install + compiled-check happen exactly once per process.

710 if _cython_module is None:

711 ok, msg = _try_load_cython_package()

712 if not ok:

713 logger.debug(f"[try_import_hevolveai] hevolveai unavailable: {msg}")

714 return None

715

716 if dotted_path == 'hevolveai':

717 return _cython_module

718

719 try:

720 import importlib

721 return importlib.import_module(dotted_path)

722 except ImportError as e:

723 logger.debug(f"[try_import_hevolveai] {dotted_path}: {e}")

724 return None

725 except Exception as e:

726 logger.debug(f"[try_import_hevolveai] {dotted_path} unexpected: {e}")

727 return None

728

729

730def try_import_hevolveai_names(

731 dotted_path: str,

732 names: Tuple[str, ...],

733) -> Optional[Tuple[Any, ...]]:

734 """Import a hevolveai submodule and extract specific attributes.

735

736 Convenience wrapper for the common ``from X.Y import A, B`` pattern.

737 Returns a tuple of attribute values, or ``None`` if the module or

738 any named attribute is missing.

739

740 Example:

741 >>> result = try_import_hevolveai_names(

742 ... 'hevolveai.embodied_ai.utils.visual_encoding',

743 ... ('compute_frame_difference', 'decode_jpeg'),

744 ... )

745 >>> if result is None:

746 ... # use fallback

747 ... ...

748 ... else:

749 ... compute_frame_difference, decode_jpeg = result

750 """

751 mod = try_import_hevolveai(dotted_path)

752 if mod is None:

753 return None

754 try:

755 return tuple(getattr(mod, n) for n in names)

756 except AttributeError as e:

757 logger.debug(f"[try_import_hevolveai_names] {dotted_path}: {e}")

758 return None

Coverage for security / native_hive_loader.py: 29.4%

320 statements