Skip to content

AdvSteg

stegobox.codec.AdvSteg

Bases: BaseCodec

Source code in stegobox/codec/advsteg/advsteg.py
class AdvSteg(BaseCodec):
    def __init__(
        self,
        image_size: int = 109,
        output_size: int = 64,
        msg_length: int = 100,
        weights_path: str = "ckpt/advsteg/weights.pt",
        cuda: bool = False,
        verbose: bool = True,
    ) -> None:
        """Generating Steganographic Images via Adversarial Training (NIPS 2017).

        Link to paper: [Generating Steganographic Images via Adversarial Training -
        Abstract](https://papers.nips.cc/paper/2017/hash/fe2d010308a6b3799a3d9c728ee74244-Abstract.html).

        Trained on the entire CelebA dataset for 500 epochs. Pre-trained weights support
        encoding payload of up to 100 bits. To train your own model, use the `train.py`
        script. For more details:

        * Check out my reproduction of the original paper in PyTorch:
          [spencerwooo/advsteg-pytorch](https://github.com/spencerwooo/advsteg-pytorch).
        * For reference, the original author's TensorFlow implementation:
          [jhayes14/advsteg](https://github.com/jhayes14/advsteg).

        Args:
            image_size: Input image size. Defaults to 109.
            output_size: Output generated image size. Defaults to 64.
            msg_length: Length of encoded payload. Defaults to 100.
            weights_path: Path to weights. Defaults to "ckpt/advsteg/weights.pt".
            cuda: Whether to use GPU or not. Defaults to False.
            verbose: Verbose logging enabled or not. Defaults to True.
        """

        device = torch.device("cuda" if torch.cuda.is_available() and cuda else "cpu")

        self.image_size = image_size
        self.output_size = output_size
        self.msg_length = msg_length
        self.device = device
        self.verbose = verbose

        weights = torch.load(weights_path, map_location=device)
        self.alice = AliceEncoder(msg_length, image_size, output_size).to(device)
        self.bob = BobDecoder(msg_length, output_size).to(device)

        self.alice.load_state_dict(weights["anet"])
        self.bob.load_state_dict(weights["bnet"])
        self.alice.eval()
        self.bob.eval()

        if verbose:
            print(f"Loaded AdvSteg weights from {weights_path}")

    def _transform(self, image: Image.Image) -> torch.Tensor:
        transform = transforms.Compose(
            [
                transforms.Resize(self.image_size),
                transforms.CenterCrop(self.image_size),
                transforms.ToTensor(),
                transforms.Normalize((0.5, 0.5, 0.5), (0.5, 0.5, 0.5)),
            ]
        )
        return transform(image)  # type: ignore

    def _inv_transform(self, image: torch.Tensor) -> Image.Image:
        inv_transform = transforms.Compose(
            [
                transforms.Normalize((-1, -1, -1), (2, 2, 2)),
                transforms.ToPILImage(),
            ]
        )
        return inv_transform(image)  # type: ignore

    def encode(self, carrier: Image.Image, payload: str) -> Image.Image:
        """Encode payload into carrier image.

        Args:
            carrier: Carrier image in PIL format.
            payload: Payload string, up to `msg_length // 8` characters.

        Returns:
            Encoded stego image in PIL format.
        """

        cover = self._transform(carrier).unsqueeze(0).to(self.device)

        # check if payload fits in carrier
        if len(payload) * 8 > self.msg_length:
            # cut off overlengthed payload
            payload = payload[: self.msg_length // 8]
            print(f"Warning: payload too long, truncated to {self.msg_length} bits.")

        # encode payload utf-8 to binary, normalize, and convert to torch tensor
        payloadb = payload.encode("utf-8")
        payloadb = int.from_bytes(payloadb, "big")
        payloadb = bin(payloadb)[2:].zfill(self.msg_length)

        payloadb = torch.tensor([int(i) for i in payloadb]).unsqueeze(0)
        payloadb = (payloadb * 2 - 1) / 2
        payloadb = payloadb.to(self.device)

        # embed payload with alice
        stego = self.alice(cover, payloadb)
        return self._inv_transform(stego.squeeze(0))

    def decode(self, carrier: Image.Image) -> str:
        """Decode payload from carrier image.

        Args:
            carrier: Stego image in PIL format.

        Returns:
            Decoded payload string.
        """

        cover = self._transform(carrier).unsqueeze(0).to(self.device)

        # decode payload with bob
        payloadb = self.bob(cover)
        payloadb = (payloadb * 2 + 1) / 2
        payloadb = payloadb.squeeze(0).detach().cpu().numpy()
        payloadb = payloadb.round().astype(int)

        # convert binary to utf-8
        payloadb = "".join([str(i) for i in payloadb])
        payloadb = int(payloadb, 2)
        payloadb = payloadb.to_bytes((payloadb.bit_length() + 7) // 8, "big").decode()

        return payloadb

__init__(image_size=109, output_size=64, msg_length=100, weights_path='ckpt/advsteg/weights.pt', cuda=False, verbose=True)

Generating Steganographic Images via Adversarial Training (NIPS 2017).

Link to paper: Generating Steganographic Images via Adversarial Training - Abstract.

Trained on the entire CelebA dataset for 500 epochs. Pre-trained weights support encoding payload of up to 100 bits. To train your own model, use the train.py script. For more details:

Parameters:

Name Type Description Default
image_size int

Input image size. Defaults to 109.

109
output_size int

Output generated image size. Defaults to 64.

64
msg_length int

Length of encoded payload. Defaults to 100.

100
weights_path str

Path to weights. Defaults to "ckpt/advsteg/weights.pt".

'ckpt/advsteg/weights.pt'
cuda bool

Whether to use GPU or not. Defaults to False.

False
verbose bool

Verbose logging enabled or not. Defaults to True.

True
Source code in stegobox/codec/advsteg/advsteg.py
def __init__(
    self,
    image_size: int = 109,
    output_size: int = 64,
    msg_length: int = 100,
    weights_path: str = "ckpt/advsteg/weights.pt",
    cuda: bool = False,
    verbose: bool = True,
) -> None:
    """Generating Steganographic Images via Adversarial Training (NIPS 2017).

    Link to paper: [Generating Steganographic Images via Adversarial Training -
    Abstract](https://papers.nips.cc/paper/2017/hash/fe2d010308a6b3799a3d9c728ee74244-Abstract.html).

    Trained on the entire CelebA dataset for 500 epochs. Pre-trained weights support
    encoding payload of up to 100 bits. To train your own model, use the `train.py`
    script. For more details:

    * Check out my reproduction of the original paper in PyTorch:
      [spencerwooo/advsteg-pytorch](https://github.com/spencerwooo/advsteg-pytorch).
    * For reference, the original author's TensorFlow implementation:
      [jhayes14/advsteg](https://github.com/jhayes14/advsteg).

    Args:
        image_size: Input image size. Defaults to 109.
        output_size: Output generated image size. Defaults to 64.
        msg_length: Length of encoded payload. Defaults to 100.
        weights_path: Path to weights. Defaults to "ckpt/advsteg/weights.pt".
        cuda: Whether to use GPU or not. Defaults to False.
        verbose: Verbose logging enabled or not. Defaults to True.
    """

    device = torch.device("cuda" if torch.cuda.is_available() and cuda else "cpu")

    self.image_size = image_size
    self.output_size = output_size
    self.msg_length = msg_length
    self.device = device
    self.verbose = verbose

    weights = torch.load(weights_path, map_location=device)
    self.alice = AliceEncoder(msg_length, image_size, output_size).to(device)
    self.bob = BobDecoder(msg_length, output_size).to(device)

    self.alice.load_state_dict(weights["anet"])
    self.bob.load_state_dict(weights["bnet"])
    self.alice.eval()
    self.bob.eval()

    if verbose:
        print(f"Loaded AdvSteg weights from {weights_path}")

encode(carrier, payload)

Encode payload into carrier image.

Parameters:

Name Type Description Default
carrier Image

Carrier image in PIL format.

required
payload str

Payload string, up to msg_length // 8 characters.

required

Returns:

Type Description
Image

Encoded stego image in PIL format.

Source code in stegobox/codec/advsteg/advsteg.py
def encode(self, carrier: Image.Image, payload: str) -> Image.Image:
    """Encode payload into carrier image.

    Args:
        carrier: Carrier image in PIL format.
        payload: Payload string, up to `msg_length // 8` characters.

    Returns:
        Encoded stego image in PIL format.
    """

    cover = self._transform(carrier).unsqueeze(0).to(self.device)

    # check if payload fits in carrier
    if len(payload) * 8 > self.msg_length:
        # cut off overlengthed payload
        payload = payload[: self.msg_length // 8]
        print(f"Warning: payload too long, truncated to {self.msg_length} bits.")

    # encode payload utf-8 to binary, normalize, and convert to torch tensor
    payloadb = payload.encode("utf-8")
    payloadb = int.from_bytes(payloadb, "big")
    payloadb = bin(payloadb)[2:].zfill(self.msg_length)

    payloadb = torch.tensor([int(i) for i in payloadb]).unsqueeze(0)
    payloadb = (payloadb * 2 - 1) / 2
    payloadb = payloadb.to(self.device)

    # embed payload with alice
    stego = self.alice(cover, payloadb)
    return self._inv_transform(stego.squeeze(0))

decode(carrier)

Decode payload from carrier image.

Parameters:

Name Type Description Default
carrier Image

Stego image in PIL format.

required

Returns:

Type Description
str

Decoded payload string.

Source code in stegobox/codec/advsteg/advsteg.py
def decode(self, carrier: Image.Image) -> str:
    """Decode payload from carrier image.

    Args:
        carrier: Stego image in PIL format.

    Returns:
        Decoded payload string.
    """

    cover = self._transform(carrier).unsqueeze(0).to(self.device)

    # decode payload with bob
    payloadb = self.bob(cover)
    payloadb = (payloadb * 2 + 1) / 2
    payloadb = payloadb.squeeze(0).detach().cpu().numpy()
    payloadb = payloadb.round().astype(int)

    # convert binary to utf-8
    payloadb = "".join([str(i) for i in payloadb])
    payloadb = int(payloadb, 2)
    payloadb = payloadb.to_bytes((payloadb.bit_length() + 7) // 8, "big").decode()

    return payloadb