Jump to content

Connect Leeroopedia MCP: Equip your AI agents to search best practices, build plans, verify code, diagnose failures, and look up hyperparameter defaults.

Implementation:Protectai Modelscan Model

From Leeroopedia
Knowledge Sources
Domains ML_Security, Software_Architecture
Last Updated 2026-02-14 12:00 GMT

Overview

Concrete tool for wrapping model file paths and byte streams into a uniform scanning interface, provided by the modelscan model module.

Description

The Model class provides a context manager-based abstraction over model files. It wraps a file source (path or archive entry identifier) and an optional byte stream into a single object that scanners and middleware can consume uniformly. The class manages stream lifecycle (open/close), provides seekable stream access, and carries a context dictionary for middleware-attached metadata (e.g., detected format tags).

Usage

Use this class when:

  • Implementing a custom scanner that needs to read model bytes via get_stream()
  • Understanding how modelscan passes files through the middleware and scanner pipeline
  • Working with zip archive entries that are represented as Model objects with pre-opened streams

Code Reference

Source Location

  • Repository: modelscan
  • File: modelscan/model.py
  • Lines: L9-57

Signature

class Model:
    def __init__(
        self,
        source: Union[str, Path],
        stream: Optional[IO[bytes]] = None,
    ):
        """
        Args:
            source: File path or "archive:entry" identifier string.
            stream: Optional pre-opened byte stream (for zip entries).
                    If None, stream is opened from source path on __enter__.
        """

    def open(self) -> "Model":
        """Open file stream if not already provided. Returns self."""

    def close(self) -> None:
        """Close stream if opened by this instance (not for passed-in streams)."""

    def __enter__(self) -> "Model":
        """Context manager entry - calls open()."""

    def __exit__(self, exc_type, exc_value, traceback) -> None:
        """Context manager exit - calls close()."""

    def get_source(self) -> Path:
        """Return the source path identifier."""

    def get_stream(self, offset: int = 0) -> IO[bytes]:
        """
        Return byte stream seeked to offset.
        Raises ModelDataEmpty if stream is not available.
        """

    def set_context(self, key: str, value: Any) -> None:
        """Set a context metadata value (e.g., format tags)."""

    def get_context(self, key: str) -> Any:
        """Get a context metadata value."""

Import

from modelscan.model import Model

I/O Contract

Inputs

Name Type Required Description
source Union[str, Path] Yes File path for filesystem files, or "archive_path:entry_name" string for zip entries
stream Optional[IO[bytes]] No Pre-opened byte stream for zip archive entries. If None, the file is opened from source path.

Outputs

Name Type Description
get_source() Path The source path identifier as a Path object
get_stream() IO[bytes] Seekable byte stream of the file contents, positioned at the requested offset
get_context("formats") List[Property] List of SupportedModelFormats Property objects set by the middleware pipeline

Usage Examples

Basic File Usage

from modelscan.model import Model

# Use as context manager for automatic stream lifecycle
with Model("/path/to/model.pkl") as model:
    source = model.get_source()     # Path("/path/to/model.pkl")
    stream = model.get_stream()      # IO[bytes] positioned at start
    data = stream.read(100)          # Read first 100 bytes

    # Re-read from beginning
    stream = model.get_stream(offset=0)

Zip Archive Entry

import zipfile
from modelscan.model import Model

with zipfile.ZipFile("/path/to/model.pt", "r") as zf:
    for entry_name in zf.namelist():
        with zf.open(entry_name, "r") as entry_stream:
            # Create Model with pre-opened stream
            model = Model(
                source=f"/path/to/model.pt:{entry_name}",
                stream=entry_stream,
            )
            # Scanner can use model.get_source() and model.get_stream()

Using Context for Format Tags

from modelscan.model import Model
from modelscan.settings import SupportedModelFormats

with Model("/path/to/model.pkl") as model:
    # Middleware sets format context
    model.set_context("formats", [SupportedModelFormats.PICKLE])

    # Scanner checks format context
    formats = model.get_context("formats")
    if any(f.value == "pickle" for f in formats):
        # Proceed with pickle scanning
        pass

Related Pages

Implements Principle

Requires Environment

Page Connections

Double-click a node to navigate. Hold to expand connections.
Principle
Implementation
Heuristic
Environment