API Reference

This reference documents the core functions available in the top-level argus package.

Core Functions

argus.api.seal(data, schema='GENERIC', detached=False)

Seals the provided data into a Merkle Tree and returns the cryptographic receipt.

Parameters:

Name	Type	Description	Default
data	Union[List[Dict], List[str], bytes]	List of log entries (dicts/strings) or raw bytes.	required
schema	str	Schema identifier (e.g. "FINANCIAL_V1").	'GENERIC'
detached	bool	If True, returns minimal receipt without original data.	False

Returns:

Name	Type	Description
Dict	Dict	Evidence Package with 'seal', 'receipts', and optional 'data'.

Source code in argus/api.py

def seal(data: Union[List[Dict], List[str], bytes], 
         schema: str = "GENERIC", 
         detached: bool = False) -> Dict:
    """
    Seals the provided data into a Merkle Tree and returns the cryptographic receipt.

    Args:
        data: List of log entries (dicts/strings) or raw bytes.
        schema: Schema identifier (e.g. "FINANCIAL_V1").
        detached: If True, returns minimal receipt without original data.

    Returns:
        Dict: Evidence Package with 'seal', 'receipts', and optional 'data'.
    """
    # 1. Normalize & Canonicalize
    if not isinstance(data, list):
        data = [data] # Promote single item to list

    # Process: Transform dicts to strings, keep bytes/strings as is
    processed_logs = [canonicalize(item) for item in data]

    # 2. Build Tree
    tree = ArgusMerkleTree(processed_logs)
    root = tree.get_root()
    count = len(data)

    # 3. Generate Receipt
    timestamp = int(time.time() * 1_000_000_000)
    receipts_list = []

    for i in range(count):
        proof = tree.get_proof(i)
        original = data[i]

        # Store log: if bytes, base64 encode; if dict/str, keep as is
        if detached:
            stored_log = None
        elif isinstance(original, bytes):
            stored_log = f"base64:{base64.b64encode(original).decode('utf-8')}"
        else:
            stored_log = original

        receipts_list.append({
            "index": i,
            "log": stored_log,
            "receipt": proof
        })

    return {
        "seal": {
            "magic": "ARGS",
            "version": "v1.1",
            "schema": schema,
            "timestamp_iso": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(timestamp / 1e9)),
            "timestamp_int": timestamp,
            "root_hash": root,
            "count": count,
            "detached": detached
        },
        "data": None if detached else data,
        "receipts": receipts_list
    }

argus.api.verify(receipt, data=None, strict=False)

Verifies the integrity of an Evidence Package or Root Hash.

Parameters:

Name	Type	Description	Default
receipt	Dict	The Argus receipt/evidence dictionary.	required
data	Optional[Union[List, bytes]]	The original data to verify.	None
strict	bool	If True, raises ArgusVerificationError with detailed failure indices on mismatch.	False

Returns:

Name	Type	Description
bool	bool	True if authentic, False otherwise (unless strict=True).

Source code in argus/api.py

def verify(receipt: Dict, data: Optional[Union[List, bytes]] = None, strict: bool = False) -> bool:
    """
    Verifies the integrity of an Evidence Package or Root Hash.

    Args:
        receipt: The Argus receipt/evidence dictionary.
        data: The original data to verify.
        strict: If True, raises ArgusVerificationError with detailed failure indices on mismatch.

    Returns:
        bool: True if authentic, False otherwise (unless strict=True).
    """
    # 1. Extract Root & Data (Polymorphic inputs)
    sealed_root = None
    target_data = None
    receipts_list = None

    if isinstance(receipt, str): # Case: Root String
        sealed_root = receipt
        target_data = data
    elif isinstance(receipt, dict):
        sealed_root = receipt.get("seal", {}).get("root_hash") or receipt.get("root_hash")
        # Use provided data, fallback to receipt data
        target_data = data if data is not None else receipt.get("data")
        receipts_list = receipt.get("receipts")

    if not sealed_root or target_data is None:
        if strict: raise ValueError("Missing root hash or data for verification.")
        return False

    # 2. Fast Path: Recompute Root (O(N))
    if not isinstance(target_data, list):
        target_data = [target_data]

    processed_logs = [canonicalize(item) for item in target_data]
    tree = ArgusMerkleTree(processed_logs)
    calc_root = tree.get_root()

    if calc_root == sealed_root:
        return True

    # 3. Slow Path: Diagnostics (Find bad indices)
    failures = []

    # We can only diagnose if we have the individual proofs (receipts_list)
    if receipts_list and len(receipts_list) == len(target_data):
        import json
        for i, item_data in enumerate(target_data):
            # Check if we have a proof for this index
            proof_item = next((r for r in receipts_list if r["index"] == i), None)
            if not proof_item:
                continue

            # Check individual proof
            # Replicate proof verification logic
            if isinstance(item_data, (dict, list)):
                log_bytes = canonicalize(item_data).encode('utf-8')
            elif isinstance(item_data, str):
                log_bytes = item_data.encode('utf-8')
            else:
                log_bytes = item_data

            if not verify_receipt(log_bytes, proof_item['receipt'], sealed_root):
                failures.append(i)

    if strict:
        msg = "Integrity check failed."
        if failures:
            msg += f" Tamper detected at indices: {failures}"
        else:
            msg += " Root hash mismatch (unknown indices)."
        raise ArgusVerificationError(msg, failures)

    return False

Advanced API

argus.api.verify_proof(data, proof, root=None)

Statelessly verifies a single Merkle Proof.

Source code in argus/api.py

def verify_proof(data: Union[str, bytes], proof: Dict, root: Optional[str] = None) -> bool:
    """
    Statelessly verifies a single Merkle Proof.
    """
    if root is None:
        root = proof.get("batch_ref")
    if not root:
        return False

    # Use canonicalize logic but careful with bytes
    # verify_receipt expects bytes
    if isinstance(data, (dict, list)):
        log_data = canonicalize(data).encode('utf-8')
    elif isinstance(data, str):
        log_data = data.encode('utf-8')
    else:
        log_data = data

    return verify_receipt(log_data, proof['path'], root)

argus.api.proof(receipt, index)

Extracts a lightweight proof for a specific item from a Master Receipt. (Formerly extract_proof)

Source code in argus/api.py

def proof(receipt: Dict, index: int) -> Dict:
    """
    Extracts a lightweight proof for a specific item from a Master Receipt.
    (Formerly extract_proof)
    """
    receipts_list = receipt.get("receipts", [])
    target = next((r for r in receipts_list if r["index"] == index), None)

    if not target:
        raise ValueError(f"Log index {index} not found.")

    # Logic to extract the log object safely
    target_log = target["log"]
    try:
        # If it's a JSON string and not base64, try to unpack it to dict
        if isinstance(target_log, str) and not target_log.startswith("base64:") and target_log.strip().startswith("{"):
            target_data = json.loads(target_log)
        else:
            target_data = target_log
    except (json.JSONDecodeError, AttributeError):
        # Fallback if parsing fails or if not a string
        target_data = target_log

    root = receipt.get("seal", {}).get("root_hash") or receipt.get("root_hash")

    return {
        "argos_spec": "v1.0",
        "data": target_data,
        "proof": {
            "batch_ref": root,
            "path": target["receipt"],
            "index": target["index"]
        }
    }

Utilities

argus.api.init(path='.', config=None, force=False)

Initialize an Argus project. (Formerly init_project)

Source code in argus/api.py

def init(path: str = ".", config: Optional[Dict] = None, force: bool = False) -> str:
    """
    Initialize an Argus project. (Formerly init_project)
    """
    default_config = {
        "version": "1.0",
        "algorithm": "sha256",
        "input_pattern": "./logs/*.json",
        "output_receipts": "./receipts.json",
        "output_root": "./root.txt"
    }
    if config: default_config.update(config)

    target_file = os.path.join(path, "argus.config.json")
    if os.path.exists(target_file) and not force:
        raise FileExistsError(f"{target_file} exists.")

    with builtins.open(target_file, "w") as f:
        json.dump(default_config, f, indent=2)
    return target_file

argus.api.load(source, schema_config=None)

Smart load data. (Formerly load_data, replaces open to avoid shadowing)

Source code in argus/api.py

def load(source: str, schema_config: Optional[Dict] = None) -> List[Union[str, bytes]]:
    """
    Smart load data. (Formerly load_data, replaces open to avoid shadowing)
    """
    return read_logs_from_file(source, schema_config=schema_config)