论文：HiDDeN: Hiding Data With Deep Networks
作者：Jiren Zhu, Russell Kaplan, Justin Johnson, Li Fei-Fei

---

一、研究背景

在图像信息隐藏领域，通常有两类典型的应用场景：

• 隐写 (Steganography)

  • 目标：实现秘密通信。
  • 要求：消息可以从图像中被接收方解码出来，但攻击者很难区分哪些图像包含信息。
  • 关键点：隐蔽性，难以被检测。

• 数字水印 (Digital Watermarking)

  • 目标：主要用于版权保护和身份认证。
  • 要求：即便图像经过压缩、裁剪、模糊等破坏，仍能正确恢复水印信息。
  • 关键点：鲁棒性，保证信息可恢复。

传统方法多依赖人工设计特征，如：

• 修改像素的 最低有效位 (LSB)；
• 在频域的低频部分嵌入信息。

这些方法在特定场景下有效，但适应性较差。 HiDDeN 则提出了全新的思路：利用 端到端可训练的卷积神经网络 替代传统手工特征，实现更强的 灵活性 和 鲁棒性。

---

二、核心思想

HiDDeN 将数据隐藏任务设计为一个 可微分的端到端管道，通过深度学习来自动学习嵌入策略：

1. 编码器 (Encoder)

• 使用多层卷积提取封面图的特征。
• 将消息向量扩展成与图像相同空间维度的“消息体积”，与特征拼接。
• 最终生成含密图，保证与封面图在视觉上接近。

2. 噪声层 (Noise Layer)

在训练过程中引入失真，模拟现实场景：

• Dropout / Cropout：随机替换像素或区域。
• Crop：保留图像的一部分，裁剪其余部分。
• Gaussian Blur：模拟图像模糊。
• JPEG Mask / JPEG Drop：近似真实 JPEG 压缩的可微方法，保证训练过程中梯度可传播。

3. 解码器 (Decoder)

• 多层卷积提取失真图像特征。
• 使用全局平均池化获取消息相关信息。
• 最终通过线性层输出消息位。
• 解码器支持输入大小变化，因此对裁剪等操作具有适应性。

4. 对抗鉴别器 (Adversary)

• 结构类似解码器。
• 输出一个二分类概率：判断图像是否含密。
• 通过对抗训练提升含密图的隐蔽性，降低被检测概率。

---

这种设计使模型能够在容量（信息嵌入量）、隐蔽性（难以检测）、鲁棒性（抗失真能力） =三方面取得平衡。

---

三、代码实现

实现了一个简化版 HiDDeN框架在CIFAR-10上的水印嵌入实验，用于在CIFAR-10图像中嵌入和恢复二进制水印消息，并验证其在多种噪声条件下的鲁棒性，以下文件为main.py文件完整程序代码。

import argparse, math, os
import torch
import torch.nn as nn
import torch.nn.functional as F
from torch.utils.data import DataLoader
from torchvision import datasets, transforms, utils# ------------------------------
# Utils
# ------------------------------
def psnr(img1, img2, eps=1e-8):# img in [0,1]mse = F.mse_loss(img1, img2, reduction='mean').item()if mse < eps: return 99.0return 10.0 * math.log10(1.0 / mse)def make_gaussian_kernel(ks=5, sigma=1.0, device="cpu"):ax = torch.arange(ks, dtype=torch.float32) - (ks - 1) / 2.0xx, yy = torch.meshgrid(ax, ax, indexing="ij")kernel = torch.exp(-(xx**2 + yy**2) / (2 * sigma**2))kernel = kernel / kernel.sum()return kernel.to(device)# ------------------------------
# Noise layers (Identity / PixelDropout / GaussianBlur / Combined)
# ------------------------------
class NoiseLayer(nn.Module):def __init__(self, kind="identity", p=0.3, gs_ks=5, gs_sigma=1.0):super().__init__()self.kind = kindself.p = pself.gs_ks = gs_ksself.gs_sigma = gs_sigmaself.register_buffer("gs_kernel", torch.empty(0))  # init later per devicedef forward(self, ico, ien):if self.kind == "identity":return ienelif self.kind == "dropout":# pixel-wise mix: with prob p use cover pixel, else use encoded pixelif not self.training:# at test-time keep the same behavior for robustness evalpassmask = (torch.rand_like(ien[:, :1, :, :]) < self.p).float()  # [B,1,H,W]return mask * ico + (1.0 - mask) * ienelif self.kind == "gaussian":# depthwise gaussian blurif self.gs_kernel.numel() == 0 or self.gs_kernel.device != ien.device:k = make_gaussian_kernel(self.gs_ks, self.gs_sigma, ien.device)self.gs_kernel = k[None, None, :, :]  # [1,1,ks,ks]padding = self.gs_ks // 2C = ien.size(1)weight = self.gs_kernel.expand(C, 1, self.gs_ks, self.gs_ks)return F.conv2d(ien, weight, bias=None, stride=1, padding=padding, groups=C)elif self.kind == "combined":# randomly pick one per batchchoice = torch.randint(0, 3, (1,), device=ien.device).item()if choice == 0:return self.forward(ico, ien)  # combined -> fallthrough (avoid recursion)elif choice == 1:# simulate dropout with p uniformly sampled around target pp = float(torch.empty(1).uniform_(self.p*0.8, min(0.95, self.p*1.2)))mask = (torch.rand_like(ien[:, :1, :, :]) < p).float()return mask * ico + (1.0 - mask) * ienelse:# gaussian with slight sigma jittersigma = float(torch.empty(1).uniform_(max(0.5, self.gs_sigma*0.6),self.gs_sigma*1.5))ks = self.gs_ksk = make_gaussian_kernel(ks, sigma, ien.device)weight = k[None, None, :, :].expand(ien.size(1), 1, ks, ks)return F.conv2d(ien, weight, stride=1, padding=ks//2, groups=ien.size(1))else:return ien# ------------------------------
# Encoder / Decoder
# ------------------------------
class ConvBNReLU(nn.Module):def __init__(self, c_in, c_out, k=3, s=1, p=1):super().__init__()self.block = nn.Sequential(nn.Conv2d(c_in, c_out, k, s, p, bias=False),nn.BatchNorm2d(c_out),nn.ReLU(inplace=True),)def forward(self, x): return self.block(x)class Encoder(nn.Module):"""Input: cover image Ico [B,3,H,W], message bits M [B,L] in {0,1}Output: encoded image Ien [B,3,H,W] (clamped to [0,1])"""def __init__(self, L, img_ch=3, base=64):super().__init__()self.L = Lself.stem = nn.Sequential(ConvBNReLU(img_ch, base),ConvBNReLU(base, base),ConvBNReLU(base, base),ConvBNReLU(base, base),)# after concat([feat, msg_volume, image]), reduce to base then to 3self.fuse = ConvBNReLU(base + L + img_ch, base)self.to_img = nn.Conv2d(base, img_ch, kernel_size=1, stride=1, padding=0)def forward(self, ico, m_bits):B, C, H, W = ico.size()feat = self.stem(ico)# replicate bits spatially: [B,L] -> [B,L,H,W]m = m_bits.view(B, self.L, 1, 1).float().expand(B, self.L, H, W)x = torch.cat([feat, m, ico], dim=1)x = self.fuse(x)delta = self.to_img(x)ien = torch.clamp(delta, 0.0, 1.0)  # directly predict encoded image (simple & stable)return ienclass Decoder(nn.Module):"""Input: possibly noised image Ino [B,3,H,W]Output: logits over bits [B,L]"""def __init__(self, L, img_ch=3, base=64):super().__init__()self.L = Lself.body = nn.Sequential(ConvBNReLU(img_ch, base),ConvBNReLU(base, base),ConvBNReLU(base, base),ConvBNReLU(base, base),ConvBNReLU(base, base),ConvBNReLU(base, base),)self.head = nn.Sequential(ConvBNReLU(base, L),   # produce L feature maps)self.fc = nn.Linear(L, L)def forward(self, ino):x = self.body(ino)x = self.head(x)                    # [B,L,H,W]x = F.adaptive_avg_pool2d(x, 1)     # [B,L,1,1]x = x.view(x.size(0), self.L)       # [B,L]logits = self.fc(x)                 # [B,L]return logits# ------------------------------
# Training / Evaluation
# ------------------------------
def train(args):device = torch.device("cuda" if torch.cuda.is_available() and not args.cpu else "cpu")# Data: CIFAR-10 [0,1]tfm = transforms.ToTensor()train_set = datasets.CIFAR10(root="./data", train=True, download=True, transform=tfm)test_set  = datasets.CIFAR10(root="./data", train=False, download=True, transform=tfm)train_loader = DataLoader(train_set, batch_size=args.batch_size, shuffle=True, num_workers=2, pin_memory=True)test_loader  = DataLoader(test_set,  batch_size=args.batch_size, shuffle=False, num_workers=2, pin_memory=True)encoder = Encoder(L=args.L).to(device)decoder = Decoder(L=args.L).to(device)if args.noise == "identity":noise = NoiseLayer("identity")elif args.noise == "dropout":noise = NoiseLayer("dropout", p=args.drop_p)elif args.noise == "gaussian":noise = NoiseLayer("gaussian", gs_ks=args.gs_ks, gs_sigma=args.gs_sigma)else:noise = NoiseLayer("combined", p=args.drop_p, gs_ks=args.gs_ks, gs_sigma=args.gs_sigma)noise = noise.to(device)opt = torch.optim.Adam(list(encoder.parameters())+list(decoder.parameters()), lr=args.lr)bce = nn.BCEWithLogitsLoss()mse = nn.MSELoss()os.makedirs(args.out_dir, exist_ok=True)for epoch in range(1, args.epochs+1):encoder.train(); decoder.train(); noise.train()run_msg_loss = run_img_loss = run_acc = 0.0for imgs, _ in train_loader:imgs = imgs.to(device)B = imgs.size(0)# random bits per imagem_bits = torch.randint(0, 2, (B, args.L), device=device)# encodeien = encoder(imgs, m_bits)# noiseino = noise(imgs, ien)# decodelogits = decoder(ino)# lossesmsg_loss = bce(logits, m_bits.float())img_loss = mse(ien, imgs)loss = msg_loss + args.lambda_img * img_lossopt.zero_grad()loss.backward()opt.step()with torch.no_grad():pred = (torch.sigmoid(logits) > 0.5).long()acc = (pred == m_bits).float().mean().item()run_msg_loss += msg_loss.item() * Brun_img_loss += img_loss.item() * Brun_acc      += acc * Bn = len(train_loader.dataset)print(f"[Epoch {epoch}] msg_loss={run_msg_loss/n:.4f}  img_loss={run_img_loss/n:.5f}  bit_acc={run_acc/n:.4f}")# quick eval on test set + save a visualizationif epoch % args.eval_every == 0:test_bit_acc, test_psnr = evaluate(encoder, decoder, noise, test_loader, device)print(f"  -> Test bit_acc={test_bit_acc:.4f}  PSNR(cover,encoded)={test_psnr:.2f} dB")# dump a small griddump_examples(encoder, decoder, noise, test_loader, device, args.out_dir, epoch)torch.save({"encoder": encoder.state_dict(),"decoder": decoder.state_dict()}, os.path.join(args.out_dir, "ckpt.pt"))print("Training done. Checkpoints & samples saved to:", args.out_dir)@torch.no_grad()
def evaluate(encoder, decoder, noise, loader, device):encoder.eval(); decoder.eval(); noise.eval()acc_sum, psnr_sum, cnt = 0.0, 0.0, 0for imgs, _ in loader:imgs = imgs.to(device)B = imgs.size(0)m_bits = torch.randint(0, 2, (B, decoder.L), device=device)ien = encoder(imgs, m_bits)ino = noise(imgs, ien)logits = decoder(ino)pred = (torch.sigmoid(logits) > 0.5).long()acc = (pred == m_bits).float().mean().item()acc_sum += acc * Bpsnr_sum += psnr(imgs, ien) * Bcnt += Breturn acc_sum/cnt, psnr_sum/cnt@torch.no_grad()
def dump_examples(encoder, decoder, noise, loader, device, out_dir, epoch):encoder.eval(); decoder.eval(); noise.eval()imgs, _ = next(iter(loader))imgs = imgs.to(device)[:8]B = imgs.size(0)m_bits = torch.randint(0, 2, (B, decoder.L), device=device)ien = encoder(imgs, m_bits)ino = noise(imgs, ien)# save gridsutils.save_image(imgs,  os.path.join(out_dir, f"epoch{epoch:03d}_cover.png"),  nrow=4)utils.save_image(ien,   os.path.join(out_dir, f"epoch{epoch:03d}_encoded.png"), nrow=4)utils.save_image(ino,   os.path.join(out_dir, f"epoch{epoch:03d}_noised.png"),  nrow=4)# print first sample bits for sanitylogits = decoder(ino)pred = (torch.sigmoid(logits) > 0.5).long()print("[viz] sample#0 GT bits:", m_bits[0].tolist())print("[viz] sample#0 PR bits:", pred[0].tolist())def main():parser = argparse.ArgumentParser()parser.add_argument("--epochs", type=int, default=5)parser.add_argument("--batch-size", type=int, default=128)parser.add_argument("--L", type=int, default=30, help="number of bits to embed")parser.add_argument("--noise", type=str, default="combined",choices=["identity","dropout","gaussian","combined"])parser.add_argument("--drop-p", type=float, default=0.3)parser.add_argument("--gs-ks", type=int, default=5)parser.add_argument("--gs-sigma", type=float, default=1.0)parser.add_argument("--lambda-img", type=float, default=0.7, help="weight for image MSE")parser.add_argument("--lr", type=float, default=1e-3)parser.add_argument("--eval-every", type=int, default=1)parser.add_argument("--out-dir", type=str, default="runs_hidden_cifar10")parser.add_argument("--cpu", action="store_true")args = parser.parse_args()train(args)if __name__ == "__main__":main()

目录结构为：

Hidden
├── data
│   └── cifar-10-python.tar.gz
└── main.py

运行命令为：

# 仅保真，不加噪声（容量/保密取向）
python main.py --epochs 5 --noise identity# 针对像素级 Dropout 的鲁棒水印训练
python main.py --epochs 5 --noise dropout --drop-p 0.3# 混合扰动（Dropout/高斯 模糊 随机采样），更通用
python main.py --epochs 5 --noise combined

四、关键代码与论文公式对应

解析论文的核心公式，并将其与 main.py 代码实现逐一对应。

---

1. Encoder 对应公式

论文公式 (1)：
$$I_{en}=E_{\theta }(I_{co},M)$$

• 含义：Encoder $E_{\theta }$ 接收原始图像 $I_{co}$ 与比特消息 $M$，输出带水印的图像 $I_{en}$.

• 代码对应：

ien = encoder(imgs, m_bits)   # (公式1)

---

2. 噪声层对应公式

论文公式 (2)：
$$I_{no}=N(I_{co},I_{en})$$

• 含义：在 Encoder 和 Decoder 之间插入噪声层 $N$，输出失真后的图像 $I_{no}$.

• 代码对应：

ino = noise(imgs, ien)   # (公式2)

---

3. Decoder 对应公式

论文公式 (3)：
$$M^{\prime }=D_{\varphi }(I_{no})$$

• 含义：Decoder $D_{\varphi }$ 接收 $I_{no}$，输出预测消息 $M^{\prime }$.

• 代码对应：

logits = decoder(ino)   # (公式3)

---

4. 损失函数对应公式

论文总损失公式(4)：
$$\mathcal{L}={\lambda }_I\cdot {\mathcal{L}}_I(I_{co},I_{en})+{\lambda }_M\cdot {\mathcal{L}}_M(M,M^{\prime })+{\lambda }_A\cdot {\mathcal{L}}_A$$

• 代码对应：

msg_loss = bce(logits, m_bits.float())   # L_M
img_loss = mse(ien, imgs)                # L_I
loss = msg_loss + args.lambda_img * img_loss  # 总损失 (λ_A=0)

---

5. 评价指标对应公式

• Bit Accuracy：
  $$Acc=\frac{1}{L}\sum _{i=1}^{L}1(M_i=M_i^{\prime })$$
  代码：

acc = (pred == m_bits).float().mean().item()

• PSNR：
  $$\text{PSNR}(I_{co},I_{en})=10\cdot {\log }_{10}(1/\text{MSE}(I_{co},I_{en}))$$

代码：

def psnr(img1, img2, eps=1e-8):mse = F.mse_loss(img1, img2).item()return 10.0 * math.log10(1.0 / mse)

---

• 公式 (1) $I_{en}=E_{\theta }(I_{co},M)$ → encoder.forward
• 公式 (2) $I_{no}=N(I_{co},I_{en})$ → noise.forward
• 公式 (3) $M^{\prime }=D_{\varphi }(I_{no})$ → decoder.forward
• 公式 (4) 总损失 → msg_loss + λ * img_loss
• 指标 Bit Accuracy & PSNR → acc, psnr()