Trust Manager

biopro_sdk.host.trust_manager

Trust Architecture Manager for BioPro.

Handles cryptographic integrity verification for plugins using Ed25519 signatures and Split-Manifest architecture, including double-signing consensus verification (Signing RBAC) and covert backdoor audits.

TrustManager

Manages the verification of plugin integrity and authorship.

Source code in src/biopro_sdk/host/trust_manager.py

class TrustManager:
    """Manages the verification of plugin integrity and authorship."""

    # Common system noise and runtime output files that are implicitly ignored
    IGNORE_LIST = {
        ".DS_Store",
        "Thumbs.db",
        "__pycache__",
        ".git",
        ".github",
        ".vscode",
        ".idea",
        ".pytest_cache",
        ".venv",
        "venv",
        "cache",
        "results",
        "temp",
        "logs",
        "output",
        "dist",
        "build",
        "signature.bin",
        "project_signature.bin",
        "trust_chain.json",
        "pyproject.toml",
        "security.json",
        "dev_cert.bin",
    }

    # Files that MUST be signed and are never ignored in standard zones
    MANDATORY_EXTENSIONS = {".py", ".pyw", ".json", ".yml", ".yaml", ".fcs", ".png", ".jpg"}

    def __init__(self, root_public_key: ed25519.Ed25519PublicKey | None = None):
        """Initialize the manager with a Root Public Key and local overrides."""
        self.primary_root = root_public_key
        self.trusted_roots: list[ed25519.Ed25519PublicKey] = []
        self._load_all_roots()

        self.overrides = LocalTrustRegistry()
        self._cache = None

    def _load_all_roots(self):
        """Load all trusted root keys including those in ~/.biopro/trusted_roots/."""
        self.trusted_roots = []
        if self.primary_root:
            self.trusted_roots.append(self.primary_root)
        else:
            try:
                hardcoded_root = ed25519.Ed25519PublicKey.from_public_bytes(bytes.fromhex(BIOPRO_ROOT_PUBLIC_KEY_HEX))
                self.trusted_roots.append(hardcoded_root)
            except Exception as e:
                logger.error(f"Failed to load hardcoded root key: {e}")

        roots_dir = Path.home() / ".biopro" / "trusted_roots"
        roots_dir.mkdir(parents=True, exist_ok=True)

        for key_file in roots_dir.glob("*.pub"):
            try:
                with open(key_file, "rb") as f:
                    key_data = f.read()
                    if b"BEGIN PUBLIC KEY" in key_data:
                        pub_key = serialization.load_pem_public_key(key_data)
                    else:
                        pub_key = ed25519.Ed25519PublicKey.from_public_bytes(key_data)

                    if isinstance(pub_key, ed25519.Ed25519PublicKey):
                        self.trusted_roots.append(pub_key)
            except Exception as e:
                logger.error(f"Failed to load trusted anchor {key_file.name}: {e}")

    def trust_developer(self, developer_name: str, public_key_hex: str) -> bool:
        """Manually accepts a developer as a personal trust anchor."""
        try:
            pub_bytes = bytes.fromhex(public_key_hex)
            if len(pub_bytes) != 32:
                return False

            roots_dir = Path.home() / ".biopro" / "trusted_roots"
            roots_dir.mkdir(parents=True, exist_ok=True)

            safe_name = developer_name.lower().replace(" ", "_")
            key_path = roots_dir / f"manual_{safe_name}.pub"

            with open(key_path, "wb") as f:
                f.write(pub_bytes)

            self._load_all_roots()
            logger.info(f"Personally accepted trust for developer: {developer_name}")

            if self._cache:
                self._cache.data.clear()
                self._cache._save()

            return True
        except Exception as e:
            logger.error(f"Failed to manually trust developer {developer_name}: {e}")
            return False

    def _get_cache(self):
        if self._cache is None:
            try:
                from .trust_storage import TrustCache

                self._cache = TrustCache()
            except ImportError:
                self._cache = None
        return self._cache

    def verify_plugin(self, plugin_path: Path) -> VerificationResult:
        """Execute the full multi-layer verification for a plugin folder."""
        debug_log_path = Path.home() / ".biopro" / "trust_debug.log"
        try:
            with open(debug_log_path, "a") as log:
                log.write(f"\nComparing {plugin_path.name}...\n")
        except Exception:
            pass

        try:
            cache = self._get_cache()
            if cache and cache.is_trusted(plugin_path):
                cached_path = cache.data.get(plugin_path.name, {}).get("trust_path")
                return VerificationResult(success=True, trust_level="verified_cache", trust_path=cached_path)

            # 1. Authenticity: Double-Signing & RBAC Checks
            auth_result = self._verify_signatures(plugin_path)

            # 2. Integrity & Covert Backdoor Auditing
            integrity_result = self._check_integrity(plugin_path)

            if not auth_result.success or not integrity_result.success:
                hashes = integrity_result.calculated_hashes or {}
                if self.overrides.is_locally_trusted(plugin_path.name, hashes):
                    return VerificationResult(
                        success=True,
                        trust_level="verified_local",
                        trust_path=[
                            {"name": "Local System User", "status": "root", "key": "Manual Override"},
                            {"name": "Security Check Bypassed", "status": "anchor", "key": "Local Machine Policy"},
                        ],
                    )

                final_res = auth_result if not auth_result.success else integrity_result
                final_res.calculated_hashes = hashes

                try:
                    with open(debug_log_path, "a") as log:
                        log.write(f"Auth Failed: {final_res.error_message}\n")
                except Exception:
                    pass

                return final_res

            if cache:
                cache.mark_as_trusted(plugin_path, trust_path=auth_result.trust_path)

            return VerificationResult(
                success=True,
                trust_level="verified_developer",
                calculated_hashes=integrity_result.calculated_hashes,
                trust_path=auth_result.trust_path,
            )

        except Exception as e:
            logger.exception(f"Verification failed critically for {plugin_path.name}")
            return VerificationResult(
                success=False,
                error_message=f"Critical Verification Error: {str(e)}",
                calculated_hashes=integrity_result.calculated_hashes if "integrity_result" in locals() else None,
            )

    def _verify_signatures(self, plugin_path: Path) -> VerificationResult:
        """Verifies multi-level trust chains, double-signing, and Signing RBAC."""
        chain_file = plugin_path / "trust_chain.json"
        sig_file = plugin_path / "signature.bin"
        project_sig_file = plugin_path / "project_signature.bin"
        manifest_file = plugin_path / "pyproject.toml"
        security_file = plugin_path / "security.json"

        if not chain_file.exists():
            return VerificationResult(
                success=False,
                error_message="Security Error: trust_chain.json missing. Plugin is untrusted.",
            )
        if not sig_file.exists() or not security_file.exists():
            return VerificationResult(
                success=False,
                error_message="Security Error: signature.bin or security.json missing.",
            )

        try:
            # 1. Parse & Verify manifest and security ledger Split-Manifest bindings
            parser = ManifestParser()
            manifest_data = parser.parse_file(manifest_file)

            sec_parser = SecurityParser()
            security_data = sec_parser.parse_file(str(security_file))

            sec_parser.verify_manifest_binding(manifest_file, security_data)

            if manifest_data.get("id") != plugin_path.name:
                return VerificationResult(
                    success=False,
                    error_message="Identity Mismatch: Plugin ID does not match folder name.",
                )

            # 2. Load and verify trust links recursively
            chain = TrustChain.from_file(chain_file)
            if not chain or not chain.links:
                return VerificationResult(success=False, error_message="Invalid or empty trust chain.")

            verified_dev_keys = {}
            verified_links = []

            # Map out recognized and verified links
            for i in range(len(chain.links)):
                link = chain.links[i]
                sub_bytes = bytes.fromhex(link.subject_pub)
                sig_bytes = bytes.fromhex(link.signature)

                has_parent_delegation = False
                if i + 1 < len(chain.links):
                    next_link = chain.links[i + 1]
                    if link.issuer_name == next_link.subject_name:
                        has_parent_delegation = True
                        parent_key = ed25519.Ed25519PublicKey.from_public_bytes(bytes.fromhex(next_link.subject_pub))
                        try:
                            parent_key.verify(sig_bytes, sub_bytes)
                            verified_dev_keys[link.subject_pub] = link.subject_name
                            verified_links.append(
                                {"name": link.subject_name, "status": "verified", "key": link.subject_pub}
                            )
                        except InvalidSignature:
                            return VerificationResult(
                                success=False,
                                error_message=f"Broken Trust Link: {next_link.subject_name} signature for {link.subject_name} is invalid.",
                            )

                if not has_parent_delegation:
                    anchor_found = False
                    for root_key in self.trusted_roots:
                        try:
                            root_key.verify(sig_bytes, sub_bytes)
                            anchor_found = True
                            break
                        except Exception:
                            continue

                    if not anchor_found:
                        # Check direct trust
                        for root_key in self.trusted_roots:
                            try:
                                root_bytes = root_key.public_bytes(
                                    encoding=serialization.Encoding.Raw,
                                    format=serialization.PublicFormat.Raw,
                                )
                                if root_bytes == sub_bytes:
                                    anchor_found = True
                                    break
                            except Exception:
                                continue

                    if anchor_found:
                        verified_dev_keys[link.subject_pub] = link.subject_name
                        verified_links.append({"name": link.subject_name, "status": "anchor", "key": link.subject_pub})
                    else:
                        return VerificationResult(
                            success=False,
                            error_message=f"Untrusted Root: {link.subject_name} is not signed by a recognized BioPro Authority.",
                            developer_name=link.subject_name,
                            developer_key=link.subject_pub,
                        )

            # Verify Leaf Developer signature on security.json canonical bytes
            canonical_bytes = json.dumps(security_data, sort_keys=True, separators=(",", ":")).encode("utf-8")

            dev_link = chain.links[0]
            if dev_link.subject_pub not in verified_dev_keys:
                return VerificationResult(
                    success=False,
                    error_message=f"Untrusted Root: {dev_link.subject_name} is not signed by a recognized BioPro Authority.",
                    developer_name=dev_link.subject_name,
                    developer_key=dev_link.subject_pub,
                )

            dev_public_key = ed25519.Ed25519PublicKey.from_public_bytes(bytes.fromhex(dev_link.subject_pub))
            dev_sig = sig_file.read_bytes()
            try:
                dev_public_key.verify(dev_sig, canonical_bytes)
            except InvalidSignature:
                return VerificationResult(
                    success=False,
                    error_message="Invalid Plugin Signature (Developer check failed).",
                )

            # 3. Enforce Signing RBAC and Consensus
            required_cosigners = []
            for author in manifest_data.get("authors", []):
                if "sign_code" in author.get("permissions", []):
                    required_cosigners.append(author["name"])

            verified_names = {link.subject_name for link in chain.links if link.subject_pub in verified_dev_keys}
            for cosigner in required_cosigners:
                if cosigner not in verified_names:
                    return VerificationResult(
                        success=False,
                        error_message=f"Co-Author Untrusted: Required co-signer '{cosigner}' has not signed or is untrusted.",
                    )

            # 4. Secondary CI/CD Double-Signature verification (if present)
            if project_sig_file.exists():
                project_link = chain.links[-1]
                project_public_key = ed25519.Ed25519PublicKey.from_public_bytes(bytes.fromhex(project_link.subject_pub))
                project_sig = project_sig_file.read_bytes()
                try:
                    project_public_key.verify(project_sig, canonical_bytes)
                except InvalidSignature:
                    return VerificationResult(
                        success=False,
                        error_message="Invalid Project CI Co-Signature (Project check failed).",
                    )

            full_path = [{"name": "BioPro Core", "status": "root"}] + list(reversed(verified_links))
            return VerificationResult(success=True, trust_path=full_path)

        except Exception as e:
            return VerificationResult(success=False, error_message=f"Chain Verification Error: {str(e)}")

    def _check_integrity(self, plugin_path: Path) -> VerificationResult:
        """Verifies every file against the signed security hashes and performs covert backdoor audits."""
        try:
            security_file = plugin_path / "security.json"
            if not security_file.exists():
                return VerificationResult(success=False, error_message="Missing security.json ledger.")

            sec_parser = SecurityParser()
            security_data = sec_parser.parse_file(str(security_file))

            signed_hashes = security_data.get("hashes", {})
            raw_exclusions = security_data.get("exclusions", [])
            custom_exclusions = {e.rstrip("/") for e in raw_exclusions}

            active_ignore = self.IGNORE_LIST | custom_exclusions

            found_files = set()
            found_hashes: dict[str, str] = {}
            integrity_passed = True
            error_msg = ""

            for root, dirs, files in os.walk(plugin_path):
                # Prune standard development and system virtual environments to avoid scanning overhead and false backdoor triggers
                PRUNE_DIRS = {
                    ".venv",
                    "venv",
                    ".git",
                    "__pycache__",
                    ".idea",
                    ".vscode",
                    ".pytest_cache",
                    ".github",
                }
                dirs[:] = [d for d in dirs if d not in PRUNE_DIRS]
                # Process every single file to run covert backdoor audit checks
                for file in files:
                    rel_path = os.path.relpath(os.path.join(root, file), plugin_path)

                    # Determine if this file falls within any active ignore directories or list
                    is_ignored = False
                    path_parts = rel_path.split(os.sep)
                    for part in path_parts:
                        if part in active_ignore:
                            is_ignored = True
                            break
                    if file in active_ignore:
                        is_ignored = True

                    # 1. Smart Covert Backdoor Audit: Detect unauthorized Python or script files hiding in ignored folders
                    if is_ignored:
                        executable_exts = {".py", ".pyw", ".sh", ".exe", ".bat"}
                        if any(file.endswith(ext) for ext in executable_exts):
                            if file not in ["signature.bin", "project_signature.bin", "dev_cert.bin"]:
                                return VerificationResult(
                                    success=False,
                                    error_message=f"Unauthorized Executable in Excluded Directory: Found '{rel_path}' inside an ignored zone.",
                                    calculated_hashes=found_hashes,
                                )
                        continue

                    # Skip cryptographic verification assets
                    if file in [
                        "signature.bin",
                        "project_signature.bin",
                        "trust_chain.json",
                        "pyproject.toml",
                        "security.json",
                    ]:
                        continue

                    found_files.add(rel_path)
                    calc_hash = self._hash_file(os.path.join(root, file))
                    found_hashes[rel_path] = calc_hash

                    # 2. Check for unauthorized executable files in standard plugin directories
                    if rel_path not in signed_hashes:
                        if any(file.endswith(ext) for ext in self.MANDATORY_EXTENSIONS):
                            integrity_passed = False
                            if not error_msg:
                                error_msg = f"Unauthorized File: {rel_path} is not in the signed security hashes."
                        continue

                    # 3. Check file hash match
                    if calc_hash != signed_hashes.get(rel_path):
                        integrity_passed = False
                        if not error_msg:
                            error_msg = f"Integrity Mismatch: {rel_path} has been tampered with."

            # 4. Check for missing files
            for signed_path in signed_hashes:
                if signed_path not in found_files:
                    integrity_passed = False
                    if not error_msg:
                        error_msg = f"Missing File: {signed_path} was signed but is not present on disk."

            return VerificationResult(
                success=integrity_passed,
                trust_level="verified_developer" if integrity_passed else "untrusted",
                error_message=error_msg,
                calculated_hashes=found_hashes,
            )

        except Exception as e:
            return VerificationResult(success=False, error_message=f"Integrity Check Error: {str(e)}")

    def _hash_file(self, full_path: str) -> str:
        """Utility to calculate SHA-256 for a file."""
        hasher = hashlib.sha256()
        with open(full_path, "rb") as f:
            for chunk in iter(lambda: f.read(4096), b""):
                hasher.update(chunk)
        return hasher.hexdigest()

__init__(root_public_key=None)

Initialize the manager with a Root Public Key and local overrides.

Source code in src/biopro_sdk/host/trust_manager.py

def __init__(self, root_public_key: ed25519.Ed25519PublicKey | None = None):
    """Initialize the manager with a Root Public Key and local overrides."""
    self.primary_root = root_public_key
    self.trusted_roots: list[ed25519.Ed25519PublicKey] = []
    self._load_all_roots()

    self.overrides = LocalTrustRegistry()
    self._cache = None

trust_developer(developer_name, public_key_hex)

Manually accepts a developer as a personal trust anchor.

Source code in src/biopro_sdk/host/trust_manager.py

def trust_developer(self, developer_name: str, public_key_hex: str) -> bool:
    """Manually accepts a developer as a personal trust anchor."""
    try:
        pub_bytes = bytes.fromhex(public_key_hex)
        if len(pub_bytes) != 32:
            return False

        roots_dir = Path.home() / ".biopro" / "trusted_roots"
        roots_dir.mkdir(parents=True, exist_ok=True)

        safe_name = developer_name.lower().replace(" ", "_")
        key_path = roots_dir / f"manual_{safe_name}.pub"

        with open(key_path, "wb") as f:
            f.write(pub_bytes)

        self._load_all_roots()
        logger.info(f"Personally accepted trust for developer: {developer_name}")

        if self._cache:
            self._cache.data.clear()
            self._cache._save()

        return True
    except Exception as e:
        logger.error(f"Failed to manually trust developer {developer_name}: {e}")
        return False

verify_plugin(plugin_path)

Execute the full multi-layer verification for a plugin folder.

Source code in src/biopro_sdk/host/trust_manager.py

def verify_plugin(self, plugin_path: Path) -> VerificationResult:
    """Execute the full multi-layer verification for a plugin folder."""
    debug_log_path = Path.home() / ".biopro" / "trust_debug.log"
    try:
        with open(debug_log_path, "a") as log:
            log.write(f"\nComparing {plugin_path.name}...\n")
    except Exception:
        pass

    try:
        cache = self._get_cache()
        if cache and cache.is_trusted(plugin_path):
            cached_path = cache.data.get(plugin_path.name, {}).get("trust_path")
            return VerificationResult(success=True, trust_level="verified_cache", trust_path=cached_path)

        # 1. Authenticity: Double-Signing & RBAC Checks
        auth_result = self._verify_signatures(plugin_path)

        # 2. Integrity & Covert Backdoor Auditing
        integrity_result = self._check_integrity(plugin_path)

        if not auth_result.success or not integrity_result.success:
            hashes = integrity_result.calculated_hashes or {}
            if self.overrides.is_locally_trusted(plugin_path.name, hashes):
                return VerificationResult(
                    success=True,
                    trust_level="verified_local",
                    trust_path=[
                        {"name": "Local System User", "status": "root", "key": "Manual Override"},
                        {"name": "Security Check Bypassed", "status": "anchor", "key": "Local Machine Policy"},
                    ],
                )

            final_res = auth_result if not auth_result.success else integrity_result
            final_res.calculated_hashes = hashes

            try:
                with open(debug_log_path, "a") as log:
                    log.write(f"Auth Failed: {final_res.error_message}\n")
            except Exception:
                pass

            return final_res

        if cache:
            cache.mark_as_trusted(plugin_path, trust_path=auth_result.trust_path)

        return VerificationResult(
            success=True,
            trust_level="verified_developer",
            calculated_hashes=integrity_result.calculated_hashes,
            trust_path=auth_result.trust_path,
        )

    except Exception as e:
        logger.exception(f"Verification failed critically for {plugin_path.name}")
        return VerificationResult(
            success=False,
            error_message=f"Critical Verification Error: {str(e)}",
            calculated_hashes=integrity_result.calculated_hashes if "integrity_result" in locals() else None,
        )

VerificationResult dataclass

Result of a plugin trust verification check.

Source code in src/biopro_sdk/host/trust_manager.py

@dataclass
class VerificationResult:
    """Result of a plugin trust verification check."""

    success: bool
    trust_level: str = "untrusted"
    error_message: str = ""
    trust_path: list[dict] | None = None
    developer_name: str | None = None
    developer_key: str | None = None
    calculated_hashes: dict | None = None

biopro_sdk.host.sign_plugin

Developer Utility for Signing BioPro Plugins (Unified V2 Engine).

Commands

init: Generate a new Ed25519 developer key pair. sign: Calculate integrity hashes and sign a plugin (split-manifest). project-sign: Co-sign a plugin's security ledger as the institutional Project CI runner. registry: Export the JSON snippet for the central registry.

PluginSigner

Source code in src/biopro_sdk/host/sign_plugin.py

class PluginSigner:
    def __init__(self):
        self.dev_dir = Path.home() / ".biopro" / "dev_keys"
        self.private_key_path = self.dev_dir / "private.key"
        self.public_key_path = self.dev_dir / "public.pub"
        self.delegation_path = self.dev_dir / "delegation.json"  # Your credentials from authority

    def init_identity(self):
        """Generates a new Ed25519 identity."""
        if self.private_key_path.exists():
            logger.info("Identity already exists. Delete ~/.biopro/dev_keys/ to regenerate.")
            return

        self.dev_dir.mkdir(parents=True, exist_ok=True)
        private_key = ed25519.Ed25519PrivateKey.generate()
        public_key = private_key.public_key()

        # Save Private Key (PKCS8)
        with open(self.private_key_path, "wb") as f:
            f.write(
                private_key.private_bytes(
                    encoding=serialization.Encoding.PEM,
                    format=serialization.PrivateFormat.PKCS8,
                    encryption_algorithm=serialization.NoEncryption(),
                )
            )

        # Save Public Key (Raw Bytes for the Registry)
        with open(self.public_key_path, "wb") as f:
            f.write(public_key.public_bytes(encoding=serialization.Encoding.Raw, format=serialization.PublicFormat.Raw))

        logger.info("Identity generated successfully.")
        logger.info(f"Private Key: {self.private_key_path}")
        logger.info(f"Public Key:  {self.public_key_path}")
        logger.warning("Keep your private.key SAFE and SECRET!")

    def load_private_key(self) -> ed25519.Ed25519PrivateKey:
        if not self.private_key_path.exists():
            raise FileNotFoundError("Developer identity not found. Run 'init' first.")
        with open(self.private_key_path, "rb") as f:
            key = serialization.load_pem_private_key(f.read(), password=None)
            if not isinstance(key, ed25519.Ed25519PrivateKey):
                raise TypeError("Key is not a valid Ed25519PrivateKey")
            return key

    def sign_plugin(self, plugin_path: Path):
        """Generates security.json with manifest binding hash and creates signature.bin."""
        private_key = self.load_private_key()
        public_key = private_key.public_key()

        manifest_file = plugin_path / "pyproject.toml"
        if not manifest_file.exists():
            raise FileNotFoundError(f"pyproject.toml not found in {plugin_path}")

        # Parse & Validate manifest using new Split-Manifest strict rules
        parser = ManifestParser()
        try:
            manifest = parser.parse_file(manifest_file)
        except Exception as e:
            raise ValueError(f"Failed manifest parsing: {e}") from e

        plugin_id = manifest.get("id")
        if not plugin_id:
            raise ValueError("Plugin ID is required in manifest.")

        logger.info(f"Hashing files for {plugin_id}...")
        custom_excl = {e.rstrip("/") for e in (manifest.get("custom_exclusions") or [])}
        active_ignore = IGNORE_LIST | custom_excl
        hashes = {}
        for root, dirs, files in os.walk(plugin_path):
            dirs[:] = [d for d in dirs if d not in active_ignore]

            for file in sorted(files):
                if file in active_ignore:
                    continue

                rel_path = os.path.relpath(os.path.join(root, file), plugin_path)

                # Check path parts too, similar to TrustManager
                path_parts = rel_path.split(os.sep)
                if any(part in active_ignore for part in path_parts):
                    continue

                if any(file.endswith(ext) for ext in MANDATORY_EXTENSIONS):
                    hashes[rel_path] = self._hash_file(Path(root) / file)

        # Calculate Manifest Hash Binding (Pristine pyproject.toml exactly as written on disk)
        manifest_bytes = manifest_file.read_bytes()
        manifest_hash = hashlib.sha256(manifest_bytes).hexdigest()

        # Build security.json (Support custom exclusions if specified in the manifest)
        security_data = {
            "security_version": 1,
            "plugin_id": plugin_id,
            "manifest_hash": manifest_hash,
            "exclusions": manifest.get("custom_exclusions") or [],
            "hashes": hashes,
        }

        # Write security.json
        security_file = plugin_path / "security.json"
        with open(security_file, "w", encoding="utf-8") as f:
            json.dump(security_data, f, indent=4)

        # Sign security.json canonical bytes (preventing keys sorting/whitespaces variances)
        canonical_bytes = json.dumps(security_data, sort_keys=True, separators=(",", ":")).encode("utf-8")
        signature = private_key.sign(canonical_bytes)

        # Write signature.bin
        with open(plugin_path / "signature.bin", "wb") as f:
            f.write(signature)

        # Write trust_chain.json
        pub_bytes = public_key.public_bytes(encoding=serialization.Encoding.Raw, format=serialization.PublicFormat.Raw)

        authors = manifest.get("authors", [])
        dev_name = authors[0].get("name", "Unknown Developer") if authors else "Unknown Developer"

        dev_link = TrustLink(
            subject_name=dev_name,
            subject_pub=pub_bytes.hex(),
            issuer_name="Unknown",
            signature="0" * 128,
        )

        links = [dev_link]

        # Add delegations if present
        if self.delegation_path.exists():
            try:
                parent_chain = TrustChain.from_file(self.delegation_path)
                if parent_chain:
                    dev_link.issuer_name = parent_chain.links[0].subject_name
                    links = parent_chain.links
            except Exception as e:
                logger.warning(f"Failed to load delegation chain: {e}")

        chain = TrustChain(links=links)
        with open(plugin_path / "trust_chain.json", "w") as f:
            f.write(chain.to_json())

        logger.info(f"Successfully signed {plugin_id}")
        logger.info("Generated: signature.bin, trust_chain.json")

    def project_sign_plugin(self, plugin_path: Path, project_private_key_pem: bytes):
        """Verifies developer signatures and file integrity, then co-signs security.json as the Project CI runner."""
        security_file = plugin_path / "security.json"
        signature_file = plugin_path / "signature.bin"
        trust_file = plugin_path / "trust_chain.json"
        manifest_file = plugin_path / "pyproject.toml"

        if not security_file.exists() or not signature_file.exists() or not trust_file.exists():
            logger.error("Developer signature (signature.bin) or security ledger is missing. Rejecting pipeline.")
            return

        try:
            # 1. Parse security.json
            sec_parser = SecurityParser()
            security_data = sec_parser.parse_file(str(security_file))

            # 2. Verify Manifest cryptographic binding hash
            sec_parser.verify_manifest_binding(manifest_file, security_data)

            # 3. Verify developer Ed25519 signature
            chain = TrustChain.from_file(trust_file)
            if not chain or not chain.links:
                raise SecurityValidationError("Invalid trust chain file.")

            dev_pub_hex = chain.links[0].subject_pub
            dev_pub_key = ed25519.Ed25519PublicKey.from_public_bytes(bytes.fromhex(dev_pub_hex))

            dev_sig = signature_file.read_bytes()
            canonical_bytes = json.dumps(security_data, sort_keys=True, separators=(",", ":")).encode("utf-8")
            dev_pub_key.verify(dev_sig, canonical_bytes)

            # 4. Audit all file hashes to prevent post-signature developer tampering in pipeline
            for rel_path, expected_hash in security_data["hashes"].items():
                file_path = plugin_path / rel_path
                if not file_path.exists() or self._hash_file(file_path) != expected_hash:
                    raise SecurityValidationError(f"File {rel_path} integrity hash has changed.")

        except Exception as e:
            logger.error(f"Security validation failed before project-signing. Re-check file integrity. Error: {e}")
            return

        # 5. Load Project CI private key
        try:
            project_private_key = serialization.load_pem_private_key(project_private_key_pem, password=None)
            if not isinstance(project_private_key, ed25519.Ed25519PrivateKey):
                raise TypeError("Project key is not a valid Ed25519PrivateKey")
        except Exception as e:
            logger.error(f"Failed to load project private key: {e}")
            return

        # 6. Sign security.json canonical bytes
        project_signature = project_private_key.sign(canonical_bytes)

        # 7. Write project_signature.bin
        with open(plugin_path / "project_signature.bin", "wb") as f:
            f.write(project_signature)

        # 8. Append Project Link to trust_chain.json
        project_pub_bytes = project_private_key.public_key().public_bytes(
            encoding=serialization.Encoding.Raw, format=serialization.PublicFormat.Raw
        )

        project_link = TrustLink(
            subject_name="BioPro GitHub Actions CI",
            subject_pub=project_pub_bytes.hex(),
            issuer_name="BioPro Core Authority",
            signature=project_signature.hex(),
        )

        chain.links.append(project_link)
        with open(trust_file, "w") as f:
            f.write(chain.to_json())

        logger.info(f"Successfully applied Project signature to {security_data['plugin_id']}.")

    def delegate_identity(self, subject_pub_file: Path, subject_name: str, authority_key_path: Path | None = None):
        """Signs another developer's public key using an Authority key."""
        if authority_key_path:
            with open(authority_key_path, "rb") as f:
                key = serialization.load_pem_private_key(f.read(), password=None)
                if not isinstance(key, ed25519.Ed25519PrivateKey):
                    raise TypeError("Authority key is not a valid Ed25519PrivateKey")
                private_key = key
            with open(
                authority_key_path.with_suffix(".pub")
                if authority_key_path.with_suffix(".pub").exists()
                else self.public_key_path,
                "rb",
            ) as f:
                auth_name = "BioPro Core Authority" if "root" in str(authority_key_path).lower() else "Authority"
        else:
            private_key = self.load_private_key()
            auth_name = "Me"

        if not subject_pub_file.exists():
            logger.error(f"Subject public key file not found: {subject_pub_file}")
            return

        with open(subject_pub_file, "rb") as f:
            sub_pub_bytes = f.read()
            if len(sub_pub_bytes) != 32:
                try:
                    pub = serialization.load_pem_public_key(sub_pub_bytes)
                    sub_pub_bytes = pub.public_bytes(
                        encoding=serialization.Encoding.Raw, format=serialization.PublicFormat.Raw
                    )
                except Exception:
                    logger.error("Invalid public key format.")
                    return

        signature = private_key.sign(sub_pub_bytes)

        new_link = TrustLink(
            subject_name=subject_name,
            subject_pub=sub_pub_bytes.hex(),
            issuer_name=auth_name,
            signature=signature.hex(),
        )

        links = [new_link]

        if not authority_key_path and self.delegation_path.exists():
            parent_chain = TrustChain.from_file(self.delegation_path)
            if parent_chain:
                new_link.issuer_name = parent_chain.links[0].subject_name
                links.extend(parent_chain.links)

        chain = TrustChain(links=links)
        output_file = Path(f"delegation_{subject_name.lower().replace(' ', '_')}.json")
        with open(output_file, "w") as f:
            f.write(chain.to_json())

        logger.info(f"Delegation file created: {output_file}")

    def _hash_file(self, path: Path) -> str:
        hasher = hashlib.sha256()
        with open(path, "rb") as f:
            for chunk in iter(lambda: f.read(4096), b""):
                hasher.update(chunk)
        return hasher.hexdigest()

    def print_registry_entry(self):
        """Prints the JSON block for the registry."""
        if not self.public_key_path.exists():
            logger.error("No identity found. Run 'init' first.")
            return

        with open(self.public_key_path, "rb") as f:
            pub_hex = f.read().hex()

        entry = {"developer_id": "Your-GitHub-Username", "public_key": pub_hex}
        print("\n--- COPY THIS TO YOUR registry.json ---")
        print(json.dumps(entry, indent=4))
        print("---------------------------------------")

delegate_identity(subject_pub_file, subject_name, authority_key_path=None)

Signs another developer's public key using an Authority key.

Source code in src/biopro_sdk/host/sign_plugin.py

def delegate_identity(self, subject_pub_file: Path, subject_name: str, authority_key_path: Path | None = None):
    """Signs another developer's public key using an Authority key."""
    if authority_key_path:
        with open(authority_key_path, "rb") as f:
            key = serialization.load_pem_private_key(f.read(), password=None)
            if not isinstance(key, ed25519.Ed25519PrivateKey):
                raise TypeError("Authority key is not a valid Ed25519PrivateKey")
            private_key = key
        with open(
            authority_key_path.with_suffix(".pub")
            if authority_key_path.with_suffix(".pub").exists()
            else self.public_key_path,
            "rb",
        ) as f:
            auth_name = "BioPro Core Authority" if "root" in str(authority_key_path).lower() else "Authority"
    else:
        private_key = self.load_private_key()
        auth_name = "Me"

    if not subject_pub_file.exists():
        logger.error(f"Subject public key file not found: {subject_pub_file}")
        return

    with open(subject_pub_file, "rb") as f:
        sub_pub_bytes = f.read()
        if len(sub_pub_bytes) != 32:
            try:
                pub = serialization.load_pem_public_key(sub_pub_bytes)
                sub_pub_bytes = pub.public_bytes(
                    encoding=serialization.Encoding.Raw, format=serialization.PublicFormat.Raw
                )
            except Exception:
                logger.error("Invalid public key format.")
                return

    signature = private_key.sign(sub_pub_bytes)

    new_link = TrustLink(
        subject_name=subject_name,
        subject_pub=sub_pub_bytes.hex(),
        issuer_name=auth_name,
        signature=signature.hex(),
    )

    links = [new_link]

    if not authority_key_path and self.delegation_path.exists():
        parent_chain = TrustChain.from_file(self.delegation_path)
        if parent_chain:
            new_link.issuer_name = parent_chain.links[0].subject_name
            links.extend(parent_chain.links)

    chain = TrustChain(links=links)
    output_file = Path(f"delegation_{subject_name.lower().replace(' ', '_')}.json")
    with open(output_file, "w") as f:
        f.write(chain.to_json())

    logger.info(f"Delegation file created: {output_file}")

init_identity()

Generates a new Ed25519 identity.

Source code in src/biopro_sdk/host/sign_plugin.py

def init_identity(self):
    """Generates a new Ed25519 identity."""
    if self.private_key_path.exists():
        logger.info("Identity already exists. Delete ~/.biopro/dev_keys/ to regenerate.")
        return

    self.dev_dir.mkdir(parents=True, exist_ok=True)
    private_key = ed25519.Ed25519PrivateKey.generate()
    public_key = private_key.public_key()

    # Save Private Key (PKCS8)
    with open(self.private_key_path, "wb") as f:
        f.write(
            private_key.private_bytes(
                encoding=serialization.Encoding.PEM,
                format=serialization.PrivateFormat.PKCS8,
                encryption_algorithm=serialization.NoEncryption(),
            )
        )

    # Save Public Key (Raw Bytes for the Registry)
    with open(self.public_key_path, "wb") as f:
        f.write(public_key.public_bytes(encoding=serialization.Encoding.Raw, format=serialization.PublicFormat.Raw))

    logger.info("Identity generated successfully.")
    logger.info(f"Private Key: {self.private_key_path}")
    logger.info(f"Public Key:  {self.public_key_path}")
    logger.warning("Keep your private.key SAFE and SECRET!")

print_registry_entry()

Prints the JSON block for the registry.

Source code in src/biopro_sdk/host/sign_plugin.py

def print_registry_entry(self):
    """Prints the JSON block for the registry."""
    if not self.public_key_path.exists():
        logger.error("No identity found. Run 'init' first.")
        return

    with open(self.public_key_path, "rb") as f:
        pub_hex = f.read().hex()

    entry = {"developer_id": "Your-GitHub-Username", "public_key": pub_hex}
    print("\n--- COPY THIS TO YOUR registry.json ---")
    print(json.dumps(entry, indent=4))
    print("---------------------------------------")

project_sign_plugin(plugin_path, project_private_key_pem)

Verifies developer signatures and file integrity, then co-signs security.json as the Project CI runner.

Source code in src/biopro_sdk/host/sign_plugin.py

def project_sign_plugin(self, plugin_path: Path, project_private_key_pem: bytes):
    """Verifies developer signatures and file integrity, then co-signs security.json as the Project CI runner."""
    security_file = plugin_path / "security.json"
    signature_file = plugin_path / "signature.bin"
    trust_file = plugin_path / "trust_chain.json"
    manifest_file = plugin_path / "pyproject.toml"

    if not security_file.exists() or not signature_file.exists() or not trust_file.exists():
        logger.error("Developer signature (signature.bin) or security ledger is missing. Rejecting pipeline.")
        return

    try:
        # 1. Parse security.json
        sec_parser = SecurityParser()
        security_data = sec_parser.parse_file(str(security_file))

        # 2. Verify Manifest cryptographic binding hash
        sec_parser.verify_manifest_binding(manifest_file, security_data)

        # 3. Verify developer Ed25519 signature
        chain = TrustChain.from_file(trust_file)
        if not chain or not chain.links:
            raise SecurityValidationError("Invalid trust chain file.")

        dev_pub_hex = chain.links[0].subject_pub
        dev_pub_key = ed25519.Ed25519PublicKey.from_public_bytes(bytes.fromhex(dev_pub_hex))

        dev_sig = signature_file.read_bytes()
        canonical_bytes = json.dumps(security_data, sort_keys=True, separators=(",", ":")).encode("utf-8")
        dev_pub_key.verify(dev_sig, canonical_bytes)

        # 4. Audit all file hashes to prevent post-signature developer tampering in pipeline
        for rel_path, expected_hash in security_data["hashes"].items():
            file_path = plugin_path / rel_path
            if not file_path.exists() or self._hash_file(file_path) != expected_hash:
                raise SecurityValidationError(f"File {rel_path} integrity hash has changed.")

    except Exception as e:
        logger.error(f"Security validation failed before project-signing. Re-check file integrity. Error: {e}")
        return

    # 5. Load Project CI private key
    try:
        project_private_key = serialization.load_pem_private_key(project_private_key_pem, password=None)
        if not isinstance(project_private_key, ed25519.Ed25519PrivateKey):
            raise TypeError("Project key is not a valid Ed25519PrivateKey")
    except Exception as e:
        logger.error(f"Failed to load project private key: {e}")
        return

    # 6. Sign security.json canonical bytes
    project_signature = project_private_key.sign(canonical_bytes)

    # 7. Write project_signature.bin
    with open(plugin_path / "project_signature.bin", "wb") as f:
        f.write(project_signature)

    # 8. Append Project Link to trust_chain.json
    project_pub_bytes = project_private_key.public_key().public_bytes(
        encoding=serialization.Encoding.Raw, format=serialization.PublicFormat.Raw
    )

    project_link = TrustLink(
        subject_name="BioPro GitHub Actions CI",
        subject_pub=project_pub_bytes.hex(),
        issuer_name="BioPro Core Authority",
        signature=project_signature.hex(),
    )

    chain.links.append(project_link)
    with open(trust_file, "w") as f:
        f.write(chain.to_json())

    logger.info(f"Successfully applied Project signature to {security_data['plugin_id']}.")

sign_plugin(plugin_path)

Generates security.json with manifest binding hash and creates signature.bin.

Source code in src/biopro_sdk/host/sign_plugin.py

def sign_plugin(self, plugin_path: Path):
    """Generates security.json with manifest binding hash and creates signature.bin."""
    private_key = self.load_private_key()
    public_key = private_key.public_key()

    manifest_file = plugin_path / "pyproject.toml"
    if not manifest_file.exists():
        raise FileNotFoundError(f"pyproject.toml not found in {plugin_path}")

    # Parse & Validate manifest using new Split-Manifest strict rules
    parser = ManifestParser()
    try:
        manifest = parser.parse_file(manifest_file)
    except Exception as e:
        raise ValueError(f"Failed manifest parsing: {e}") from e

    plugin_id = manifest.get("id")
    if not plugin_id:
        raise ValueError("Plugin ID is required in manifest.")

    logger.info(f"Hashing files for {plugin_id}...")
    custom_excl = {e.rstrip("/") for e in (manifest.get("custom_exclusions") or [])}
    active_ignore = IGNORE_LIST | custom_excl
    hashes = {}
    for root, dirs, files in os.walk(plugin_path):
        dirs[:] = [d for d in dirs if d not in active_ignore]

        for file in sorted(files):
            if file in active_ignore:
                continue

            rel_path = os.path.relpath(os.path.join(root, file), plugin_path)

            # Check path parts too, similar to TrustManager
            path_parts = rel_path.split(os.sep)
            if any(part in active_ignore for part in path_parts):
                continue

            if any(file.endswith(ext) for ext in MANDATORY_EXTENSIONS):
                hashes[rel_path] = self._hash_file(Path(root) / file)

    # Calculate Manifest Hash Binding (Pristine pyproject.toml exactly as written on disk)
    manifest_bytes = manifest_file.read_bytes()
    manifest_hash = hashlib.sha256(manifest_bytes).hexdigest()

    # Build security.json (Support custom exclusions if specified in the manifest)
    security_data = {
        "security_version": 1,
        "plugin_id": plugin_id,
        "manifest_hash": manifest_hash,
        "exclusions": manifest.get("custom_exclusions") or [],
        "hashes": hashes,
    }

    # Write security.json
    security_file = plugin_path / "security.json"
    with open(security_file, "w", encoding="utf-8") as f:
        json.dump(security_data, f, indent=4)

    # Sign security.json canonical bytes (preventing keys sorting/whitespaces variances)
    canonical_bytes = json.dumps(security_data, sort_keys=True, separators=(",", ":")).encode("utf-8")
    signature = private_key.sign(canonical_bytes)

    # Write signature.bin
    with open(plugin_path / "signature.bin", "wb") as f:
        f.write(signature)

    # Write trust_chain.json
    pub_bytes = public_key.public_bytes(encoding=serialization.Encoding.Raw, format=serialization.PublicFormat.Raw)

    authors = manifest.get("authors", [])
    dev_name = authors[0].get("name", "Unknown Developer") if authors else "Unknown Developer"

    dev_link = TrustLink(
        subject_name=dev_name,
        subject_pub=pub_bytes.hex(),
        issuer_name="Unknown",
        signature="0" * 128,
    )

    links = [dev_link]

    # Add delegations if present
    if self.delegation_path.exists():
        try:
            parent_chain = TrustChain.from_file(self.delegation_path)
            if parent_chain:
                dev_link.issuer_name = parent_chain.links[0].subject_name
                links = parent_chain.links
        except Exception as e:
            logger.warning(f"Failed to load delegation chain: {e}")

    chain = TrustChain(links=links)
    with open(plugin_path / "trust_chain.json", "w") as f:
        f.write(chain.to_json())

    logger.info(f"Successfully signed {plugin_id}")
    logger.info("Generated: signature.bin, trust_chain.json")

TrustChain dataclass

The full chain of trust for a plugin.

Source code in src/biopro_sdk/host/sign_plugin.py

@dataclass
class TrustChain:
    """The full chain of trust for a plugin."""

    links: list[TrustLink]

    def to_json(self) -> str:
        return json.dumps([link.to_dict() for link in self.links], indent=4)

    @classmethod
    def from_json(cls, json_str: str) -> "TrustChain":
        data = json.loads(json_str)
        links = [TrustLink(**item) for item in data]
        return cls(links=links)

    @classmethod
    def from_file(cls, path: Path) -> Optional["TrustChain"]:
        if not path.exists():
            return None
        try:
            with open(path) as f:
                return cls.from_json(f.read())
        except Exception:
            return None

TrustLink dataclass

A single link in the trust chain (Issuer -> Subject).

Source code in src/biopro_sdk/host/sign_plugin.py

@dataclass
class TrustLink:
    """A single link in the trust chain (Issuer -> Subject)."""

    subject_name: str
    subject_pub: str  # Hex encoded Ed25519 public key
    issuer_name: str
    signature: str  # Hex encoded signature

    def to_dict(self) -> dict:
        return {
            "subject_name": self.subject_name,
            "subject_pub": self.subject_pub,
            "issuer_name": self.issuer_name,
            "signature": self.signature,
        }

sign_plugin(plugin_path, private_key_path=None, cert_path=None)

Wrapper to maintain legacy compatibility with V1 import signatures.

Source code in src/biopro_sdk/host/sign_plugin.py

def sign_plugin(plugin_path: Path, private_key_path: Path | None = None, cert_path: Path | None = None):
    """Wrapper to maintain legacy compatibility with V1 import signatures."""
    signer = PluginSigner()
    if private_key_path:
        signer.private_key_path = private_key_path
    signer.sign_plugin(plugin_path)