kiholee/TOIFAS
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Keyed masking for secure multi-user semantic communication

Browse Source 
Reproducibility package for the TIFS submission: transmit and receive
core, security stages (eavesdropper, jamming, key families, attack
difficulty, known-plaintext), real BERT token streams, closed-form
verification, and the scripts that regenerate every figure and table
from the released CSVs.
This commit is contained in:
KiHoLee
2026-08-13 21:01:32 +09:00
commit 37392bc38f
34 changed files with 2293 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
code/exp_full.py
+523
View File
@@ -0,0 +1,523 @@
"""Full-scale security evaluation for paper 11 (run under WSL CUDA).
Reuses the SSE transmit/receive core from sse_lib.py and adds an
eavesdropper receiver, a jammer channel, and structured mask families.
Main configuration d=64, P=4, Vu=16 (V=Vu^P=65,536), U=4 users, matching
the language-model token vocabulary scale.
Stages (each writes a CSV to ../data; figures come from replot_security.py
and the two result tables from make_tables.py):
A security vs SNR -> sec_snr.csv (Fig. 2)
B key length -> sec_keylen.csv (Fig. 3)
C jamming vs JSR -> sec_jam.csv (Fig. 4)
D mask families -> sec_maskfam.csv (key-family table)
E scheme comparison -> sec_compare.csv (comparison table)
F attack difficulty -> sec_sens.csv, sec_brute.csv (Figs. 5-6)
Experiment scripts write CSV only, never draw. Fixed seeds.
"""
from __future__ import annotations
import math
import numpy as np
import torch
import sse_lib as L
from sse_lib import (SSE, rayleigh_gain, snr_to_sigma2, write_csv, set_seed,
eval_ser_sse, oma_ser, DATA, DEVICE)
# ----------------------------------------------------------------------
# eavesdropper: correlate the transmitted (true-mask) frame with a
# substitute mask the eavesdropper does not truly hold.
# ----------------------------------------------------------------------
@torch.no_grad()
def eval_ser_eve(model: SSE, eve_masks: torch.Tensor, snr_list,
frames: int, chunk: int = 100_000, seed: int = 777):
model.eval().to(DEVICE)
Bn = model.unit_codebook()
true_m = model.masks()
eve_masks = eve_masks.to(DEVICE)
c = model.c
out = []
for snr_db in snr_list:
g = torch.Generator(device="cpu").manual_seed(seed + int(10 * snr_db))
err = tot = 0
for n0 in range(0, frames, chunk):
n = min(chunk, frames - n0)
digits = torch.randint(model.vu, (n, model.users, model.P),
generator=g).to(DEVICE)
e = Bn[digits] / math.sqrt(model.P)
y = (e * true_m[None, :, None, :]).sum(dim=1) / c
h = rayleigh_gain((n, model.users), device=DEVICE)
sigma = snr_to_sigma2(snr_db, model.d).to(DEVICE).sqrt()
noise = torch.randn(n, model.users, model.P, model.L, device=DEVICE)
y_rx = h[:, :, None, None] * y[:, None] + sigma * noise
r = y_rx / h[:, :, None, None].clamp_min(1e-6)
cand = Bn[None, :, :] * eve_masks[:, None, :]
scores = torch.einsum("nupl,uvl->nupv", r, cand)
wrong = (scores.argmax(-1) != digits).any(dim=2)
err += int(wrong.sum()); tot += n * model.users
out.append(err / tot)
return out
@torch.no_grad()
def eval_ser_jam(model: SSE, snr_db, jsr_db_list, frames: int,
chunk: int = 100_000, seed: int = 777, mode: str = "blind",
target: int = 0):
"""Returns the target-user SER (user `target`, the user a mask-matched
jammer aims at). The mask-matched jammer aligns with the target key,
which a mask-blind jammer cannot do. Reporting the target-user SER,
rather than the user average, isolates how efficiently each jammer can
degrade a chosen victim (Proposition 2)."""
model.eval().to(DEVICE)
Bn = model.unit_codebook()
true_m = model.masks()
c = model.c
sigma = snr_to_sigma2(snr_db, model.d).to(DEVICE).sqrt()
# matched jammer aligns with the target user's masked codeword mean
# direction (needs that user's secret key)
w_fixed = (Bn[target][None, :] * true_m[target][None, :]).repeat(model.P, 1)
w_fixed = w_fixed / w_fixed.norm()
out = []
for jsr_db in jsr_db_list:
jsr = 10.0 ** (jsr_db / 10.0)
g = torch.Generator(device="cpu").manual_seed(seed + int(10 * jsr_db))
err = tot = 0
for n0 in range(0, frames, chunk):
n = min(chunk, frames - n0)
digits = torch.randint(model.vu, (n, model.users, model.P),
generator=g).to(DEVICE)
e = Bn[digits] / math.sqrt(model.P)
y = (e * true_m[None, :, None, :]).sum(dim=1) / c
h = rayleigh_gain((n, model.users), device=DEVICE)
hJ = rayleigh_gain((n,), device=DEVICE)
if mode == "matched":
w = w_fixed[None].expand(n, model.P, model.L)
else:
w = torch.randn(n, model.P, model.L, device=DEVICE)
w = w / w.reshape(n, -1).norm(dim=1)[:, None, None].clamp_min(1e-8)
jam = (hJ * math.sqrt(jsr))[:, None, None] * w
noise = torch.randn(n, model.users, model.P, model.L, device=DEVICE)
y_rx = h[:, :, None, None] * y[:, None] + jam[:, None] + sigma * noise
r = y_rx / h[:, :, None, None].clamp_min(1e-6)
cand = Bn[None, :, :] * true_m[:, None, :]
scores = torch.einsum("nupl,uvl->nupv", r, cand)
wrong = (scores.argmax(-1) != digits).any(dim=2) # (n,U)
err += int(wrong[:, target].sum()); tot += n
out.append(err / tot)
return out
def hadamard(n: int) -> np.ndarray:
"""Sylvester construction, n a power of two."""
H = np.array([[1.0]])
while H.shape[0] < n:
H = np.block([[H, H], [H, -H]])
return H
def mean_abs_xcorr(masks: torch.Tensor) -> float:
m = masks / masks.norm(dim=1, keepdim=True).clamp_min(1e-8)
G = (m @ m.T).abs()
U = m.shape[0]
off = G[~torch.eye(U, dtype=torch.bool, device=G.device)]
return float(off.mean())
def random_mask(U, Lp):
W = torch.randn(U, Lp)
return W / W.norm(dim=1, keepdim=True) * math.sqrt(Lp)
def eve_wrong_mask(U, Lp, seed):
g = torch.Generator().manual_seed(seed)
W = torch.randn(U, Lp, generator=g)
return W / W.norm(dim=1, keepdim=True) * math.sqrt(Lp)
def get_model(P=4, vu=16, d=64, U=4, iters=4000, seed=1, freeze_W=None, tag=""):
"""Train an SSE model, optionally with fixed (frozen) masks."""
set_seed(seed)
m = SSE(P=P, vu=vu, d=d, users=U).to(DEVICE)
if freeze_W is not None:
with torch.no_grad():
m.W.copy_(freeze_W.to(DEVICE))
m.W.requires_grad_(False)
L.train_sse(m, iters=iters, batch=256, lr=3e-3, seed=seed)
m.calibrate_power()
return m
def stage_A():
print("[A] security vs SNR (V=65536) ...")
m = get_model(iters=4000)
snr = [0.0, 4.0, 8.0, 12.0, 16.0, 20.0]
frames = 800_000
legit = eval_ser_sse(m, snr, frames=frames)
ew = eve_wrong_mask(m.users, m.L, seed=20260813).to(DEVICE)
eve_w = eval_ser_eve(m, ew, snr, frames=frames)
eve_n = eval_ser_eve(m, torch.ones(m.users, m.L), snr, frames=frames)
# conventional public-mask scheme: the eavesdropper holds the same
# (public) masks and decodes exactly like a legitimate user
eve_p = eval_ser_eve(m, m.masks().detach().cpu(), snr, frames=frames)
oma = oma_ser(snr, bits=int(math.log2(m.V)))
chance = 1.0 - (1.0 / m.vu) ** m.P
write_csv(DATA / "sec_snr.csv",
["snr_db", "legit", "eve_wrong", "eve_none", "eve_public",
"oma", "chance"],
[(s, legit[i], eve_w[i], eve_n[i], eve_p[i], oma[i], chance)
for i, s in enumerate(snr)])
print(" legit:", [f"{v:.2e}" for v in legit])
print(" eve :", [f"{v:.3f}" for v in eve_w])
def train_sse_reg(m: SSE, iters=4000, batch=256, lr=3e-3, seed=1,
lam_orth=1.0, lam_flat=0.1):
"""Regularized key learning for improved spreading and de-spreading.
Adds to the digit-wise cross entropy (i) an orthogonality penalty on
the off-diagonal key Gram entries, which reduces cross-user
interference and residual leakage, and (ii) a constant-modulus
penalty that flattens the key spectrum, which maximizes the spreading
of a mask-blind jammer (Proposition 2: the jammer concentration on
candidate i is sum_k w_k^2 e_{i,k}^2 weighted through the key, and a
flat key removes any low-energy entries a jammer could exploit)."""
set_seed(seed)
m.to(DEVICE)
opt = torch.optim.Adam(m.parameters(), lr=lr)
ce = torch.nn.CrossEntropyLoss()
for it in range(1, iters + 1):
digits = torch.randint(m.vu, (batch, m.users, m.P), device=DEVICE)
snr = torch.empty(batch).uniform_(0.0, 20.0)
m.calibrate_power(8192)
scores = m(digits, snr) * m.logit_scale.exp()
loss = ce(scores.reshape(-1, m.vu), digits.reshape(-1))
mk = m.masks()
G = (mk @ mk.T) / m.L
off = G - torch.eye(m.users, device=G.device)
loss = loss + lam_orth * off.pow(2).sum()
loss = loss + lam_flat * (mk.pow(2) - 1.0).pow(2).mean()
opt.zero_grad(); loss.backward(); opt.step()
m.calibrate_power()
return m
def get_model_reg(P=4, vu=16, d=64, U=4, iters=4000, seed=1):
set_seed(seed)
m = SSE(P=P, vu=vu, d=d, users=U).to(DEVICE)
train_sse_reg(m, iters=iters, seed=seed)
return m
def stage_B():
print("[B] key length (dense grid so the curve is smooth) ...")
oma10 = oma_ser([10.0], bits=16)[0]
rows = []
for d in [16, 24, 32, 48, 64, 96, 128, 192, 256]:
m = get_model(d=d, iters=4000)
lg = eval_ser_sse(m, [10.0], frames=500_000)[0]
ew = eve_wrong_mask(m.users, m.L, seed=20260813).to(DEVICE)
ev = eval_ser_eve(m, ew, [10.0], frames=500_000)[0]
xc = mean_abs_xcorr(m.masks().detach())
rows.append((m.L, d, lg, ev, xc, oma10))
print(f" L={m.L:4d} legit={lg:.2e} eve={ev:.3f} xcorr={xc:.4f}")
write_csv(DATA / "sec_keylen.csv",
["L", "d", "legit_ser", "eve_ser", "mask_xcorr", "oma"], rows)
def stage_C():
print("[C] jamming vs JSR ...")
m = get_model(iters=4000)
jsr = [-10.0, -5.0, 0.0, 5.0, 10.0, 15.0, 20.0]
blind = eval_ser_jam(m, 10.0, jsr, frames=500_000, mode="blind", target=0)
matched = eval_ser_jam(m, 10.0, jsr, frames=500_000, mode="matched", target=0)
# target-user SER with no jammer, for the reference line
nojam = eval_ser_jam(m, 10.0, [-40.0], frames=500_000, mode="blind",
target=0)[0]
write_csv(DATA / "sec_jam.csv",
["jsr_db", "blind", "matched", "nojam"],
[(j, blind[i], matched[i], nojam) for i, j in enumerate(jsr)])
print(f" target no-jam={nojam:.2e}")
print(" blind :", [f"{v:.3f}" for v in blind])
print(" matched:", [f"{v:.3f}" for v in matched])
def stage_D():
print("[D] mask families ...")
P, vu, d, U = 4, 16, 64, 4
Lp = d // P
fams = {}
# random fixed masks
set_seed(7); fams["random"] = random_mask(U, Lp)
# Walsh-Hadamard rows (orthogonal)
Hd = torch.tensor(hadamard(Lp)[:U], dtype=torch.float32) # ||row||=sqrt(Lp)
fams["hadamard"] = Hd
rows = []
for name, W in fams.items():
m = get_model(P=P, vu=vu, d=d, U=U, iters=4000, freeze_W=W)
lg = eval_ser_sse(m, [10.0], frames=500_000)[0]
ew = eve_wrong_mask(U, Lp, seed=20260813).to(DEVICE)
ev = eval_ser_eve(m, ew, [10.0], frames=500_000)[0]
xc = mean_abs_xcorr(m.masks().detach())
rows.append((name, lg, ev, xc))
print(f" {name:9s} legit={lg:.2e} eve={ev:.3f} xcorr={xc:.4f}")
# learned masks (plain cross entropy)
m = get_model(P=P, vu=vu, d=d, U=U, iters=4000)
lg = eval_ser_sse(m, [10.0], frames=500_000)[0]
ew = eve_wrong_mask(U, Lp, seed=20260813).to(DEVICE)
ev = eval_ser_eve(m, ew, [10.0], frames=500_000)[0]
xc = mean_abs_xcorr(m.masks().detach())
rows.append(("learned", lg, ev, xc))
print(f" {'learned':9s} legit={lg:.2e} eve={ev:.3f} xcorr={xc:.4f}")
# regularized key learning (orthogonality + constant modulus)
mr = get_model_reg(P=P, vu=vu, d=d, U=U, iters=4000)
lgr = eval_ser_sse(mr, [10.0], frames=500_000)[0]
evr = eval_ser_eve(mr, ew, [10.0], frames=500_000)[0]
xcr = mean_abs_xcorr(mr.masks().detach())
rows.append(("learned_reg", lgr, evr, xcr))
print(f" {'learn_reg':9s} legit={lgr:.2e} eve={evr:.3f} xcorr={xcr:.4f}")
# jamming robustness of plain vs regularized keys (blind jammer)
jsr = [-10.0, -5.0, 0.0, 5.0, 10.0, 15.0, 20.0]
jb_plain = eval_ser_jam(m, 10.0, jsr, frames=300_000, mode="blind")
jb_reg = eval_ser_jam(mr, 10.0, jsr, frames=300_000, mode="blind")
write_csv(DATA / "sec_regjam.csv",
["jsr_db", "plain", "regularized"],
[(j, jb_plain[i], jb_reg[i]) for i, j in enumerate(jsr)])
write_csv(DATA / "sec_maskfam.csv",
["family", "legit_ser", "eve_ser", "mask_xcorr"], rows)
@torch.no_grad()
def eval_scheme(model: SSE, snr_db, frames, *, rx_masks=None, perms=None,
jam_w=None, jsr_db=None, target=0, chunk=100_000, seed=777,
decode_user=0):
"""Generic evaluator for the comparison schemes.
rx_masks: masks used at the decoding receiver (None = true masks).
perms: (U,d-index) per-user secret permutations applied at tx to
x_u; the decoder for `decode_user` inverse-permutes first.
rx side without the permutation just decodes raw.
jam_w: None or 'matched'/'blind' jammer aimed at `target`.
Returns SER of `decode_user` (frame error over its P digits)."""
model.eval().to(DEVICE)
Bn = model.unit_codebook()
true_m = model.masks()
c = model.c
sigma = snr_to_sigma2(snr_db, model.d).to(DEVICE).sqrt()
d = model.P * model.L
if perms is not None:
inv = torch.argsort(perms, dim=1)
if jam_w == "matched":
wf = (Bn[target][None, :] * true_m[target][None, :]).repeat(model.P, 1)
if perms is not None:
wfl = wf.reshape(-1)[perms[target]]
wf = wfl.reshape(model.P, model.L)
wf = wf / wf.norm()
jsr = 10.0 ** (jsr_db / 10.0) if jsr_db is not None else 0.0
g = torch.Generator(device="cpu").manual_seed(seed + int(10 * snr_db))
err = tot = 0
for n0 in range(0, frames, chunk):
n = min(chunk, frames - n0)
digits = torch.randint(model.vu, (n, model.users, model.P),
generator=g).to(DEVICE)
e = Bn[digits] / math.sqrt(model.P)
x = e * true_m[None, :, None, :] # (n,U,P,L)
if perms is not None:
xf = x.reshape(n, model.users, d)
xf = torch.stack([xf[:, u][:, perms[u]] for u in range(model.users)], 1)
x = xf.reshape(n, model.users, model.P, model.L)
y = x.sum(dim=1) / c # (n,P,L)
h = rayleigh_gain((n,), device=DEVICE) # decode_user channel
y_rx = h[:, None, None] * y # (n,P,L)
if jam_w is not None:
hJ = rayleigh_gain((n,), device=DEVICE)
if jam_w == "matched":
w = wf[None].expand(n, model.P, model.L)
else:
w = torch.randn(n, model.P, model.L, device=DEVICE)
w = w / w.reshape(n, -1).norm(dim=1)[:, None, None].clamp_min(1e-8)
y_rx = y_rx + (hJ * math.sqrt(jsr))[:, None, None] * w
y_rx = y_rx + sigma * torch.randn(n, model.P, model.L, device=DEVICE)
r = y_rx / h[:, None, None].clamp_min(1e-6)
if perms is not None:
rf = r.reshape(n, d)[:, inv[decode_user]]
r = rf.reshape(n, model.P, model.L)
m_rx = true_m if rx_masks is None else rx_masks.to(DEVICE)
cand = Bn * m_rx[decode_user][None, :] # (Vu,L)
scores = torch.einsum("npl,vl->npv", r, cand)
wrong = (scores.argmax(-1) != digits[:, decode_user]).any(dim=1)
err += int(wrong.sum()); tot += n
return err / tot
def stage_E():
"""Comparison across five schemes at 10 dB, V=65,536, user-0 metrics.
Columns: legitimate SER; outsider-eavesdropper SER; insider SER (a
curious legitimate user of the SAME system decoding user 0 with its
own credentials); target-user SER under the strongest jammer the
attacker can BUILD from public knowledge at JSR 0 dB (matched if the
masks are public, blind if the PHY structure is secret)."""
print("[E] scheme comparison ...")
m = get_model(iters=4000)
F = 400_000
d = m.P * m.L
set_seed(20260813)
ew = eve_wrong_mask(m.users, m.L, seed=20260813)
# shuffling-style multi-user adaptation: one GLOBAL secret permutation
# shared by all users (per-user permutations break the trained
# multi-user separation, so the shared key is the fair extension)
gp = torch.Generator().manual_seed(11)
gperm = torch.randperm(d, generator=gp)
perms = gperm[None].repeat(m.users, 1)
chance = 1.0 - (1.0 / m.vu) ** m.P
insider_masks = torch.roll(m.masks().detach().cpu(), 1, 0) # user 1's key
rows = []
# S1 proposed keyed masking: per-user secret masks
lg = eval_scheme(m, 10.0, F)
ev = eval_scheme(m, 10.0, F, rx_masks=ew)
ins = eval_scheme(m, 10.0, F, rx_masks=insider_masks)
jm = eval_scheme(m, 10.0, F, jam_w="blind", jsr_db=0.0)
rows.append(("proposed", lg, ev, ins, jm))
# S2 public-mask superposition (no key): everyone decodes, attacker
# builds the matched jammer
jm2 = eval_scheme(m, 10.0, F, jam_w="matched", jsr_db=0.0)
rows.append(("public_mask", lg, lg, lg, jm2))
# S3 global permutation key over public masks (shuffling-style): the
# outsider lacks the permutation, but every insider holds it and the
# masks are public, so insiders decode each other
lg3 = eval_scheme(m, 10.0, F, perms=perms)
ev3 = eval_scheme_permuted_eve(m, 10.0, F, perms)
jm3 = eval_scheme(m, 10.0, F, perms=perms, jam_w="blind", jsr_db=0.0)
rows.append(("perm_key", lg3, ev3, lg3, jm3))
# S4 per-user index cipher (one-time pad on the digits) over public
# masks: content protected from outsiders and insiders, but the PHY
# is public so the matched jammer remains buildable
rows.append(("index_cipher", lg, chance, chance, jm2))
# S5 OMA digital, no encryption: open to everyone
from sse_lib import oma_ser
lg5 = oma_ser([10.0], bits=int(math.log2(m.V)))[0]
rows.append(("oma_plain", lg5, lg5, lg5, float("nan")))
write_csv(DATA / "sec_compare.csv",
["scheme", "legit_ser", "eve_out", "eve_in", "jam0_ser"], rows)
for r in rows:
print(" ", r)
@torch.no_grad()
def eval_scheme_permuted_eve(model: SSE, snr_db, frames, perms,
chunk=100_000, seed=777):
"""Eve for S3: sees the per-user permuted tx, holds the PUBLIC masks
but not the permutation, decodes user 0 raw."""
model.eval().to(DEVICE)
Bn = model.unit_codebook()
true_m = model.masks()
c = model.c
d = model.P * model.L
sigma = snr_to_sigma2(snr_db, model.d).to(DEVICE).sqrt()
g = torch.Generator(device="cpu").manual_seed(seed + int(10 * snr_db))
err = tot = 0
for n0 in range(0, frames, chunk):
n = min(chunk, frames - n0)
digits = torch.randint(model.vu, (n, model.users, model.P),
generator=g).to(DEVICE)
e = Bn[digits] / math.sqrt(model.P)
x = e * true_m[None, :, None, :]
xf = x.reshape(n, model.users, d)
xf = torch.stack([xf[:, u][:, perms[u]] for u in range(model.users)], 1)
y = xf.reshape(n, model.users, model.P, model.L).sum(dim=1) / c
h = rayleigh_gain((n,), device=DEVICE)
y_rx = h[:, None, None] * y + sigma * torch.randn(
n, model.P, model.L, device=DEVICE)
r = y_rx / h[:, None, None].clamp_min(1e-6)
cand = Bn * true_m[0][None, :]
scores = torch.einsum("npl,vl->npv", r, cand)
wrong = (scores.argmax(-1) != digits[:, 0]).any(dim=1)
err += int(wrong.sum()); tot += n
return err / tot
def correlated_masks(true_m: torch.Tensor, rho: float, gen: torch.Generator):
"""Substitute masks with prescribed normalized correlation rho to the
true keys: mtil = rho*m + sqrt(1-rho^2)*m_perp, ||mtil|| = ||m||."""
U, Lp = true_m.shape
out = torch.empty_like(true_m)
for u in range(U):
m = true_m[u]
p = torch.randn(Lp, generator=gen)
p = p - (p @ m) / (m @ m) * m
p = p / p.norm() * m.norm()
out[u] = rho * m + math.sqrt(max(0.0, 1 - rho * rho)) * p
return out
def stage_F():
"""Attack difficulty in the style of standard security evaluations.
(i) Key sensitivity: Eve SER against the correlation rho between her
guess and the true key (avalanche-style curve).
(ii) Brute-force key search: expected Eve SER against the number of
random key guesses K, where for each trial the attacker keeps the
guess with the LARGEST correlation to the true key (a genie-aided
upper bound on any selection rule). The best-guess correlation
rho_max(K, L) is sampled by Monte Carlo and mapped through the
measured sensitivity curve of (i)."""
print("[F] attack difficulty ...")
m = get_model(iters=4000)
F = 200_000
true_m = m.masks().detach().cpu()
gen = torch.Generator().manual_seed(31)
# (i) sensitivity curve, densest where the curve falls steeply
rhos = [0.0, 0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.65, 0.7, 0.75,
0.8, 0.84, 0.88, 0.90, 0.92, 0.94, 0.96, 0.97, 0.98,
0.99, 0.995, 1.0]
sens = []
for rho in rhos:
mt = correlated_masks(true_m, rho, gen)
ser = eval_ser_eve(m, mt, [10.0], frames=F)[0]
sens.append((rho, ser))
print(f" rho={rho:.2f} eve_ser={ser:.4f}")
write_csv(DATA / "sec_sens.csv", ["rho", "eve_ser"], sens)
# (ii) brute-force: sample rho_max(K, L) and interpolate SER(rho)
import numpy as np
r_arr = np.array([r for r, _ in sens])
s_arr = np.array([s for _, s in sens])
def ser_of_rho(r):
return float(np.interp(abs(r), r_arr, s_arr))
ks = [1, 10, 100, 1_000, 10_000, 100_000, 1_000_000]
rows = []
rng = np.random.default_rng(2026)
for Lp in [8, 16, 32, 64]:
for K in ks:
trials = 400
# rho of a random unit guess vs a fixed key in R^L is the
# first coordinate of a random unit vector; sample K per trial
best = np.empty(trials)
for t in range(trials):
g = rng.standard_normal((K, Lp))
g /= np.linalg.norm(g, axis=1, keepdims=True)
best[t] = np.abs(g[:, 0]).max()
ser_est = float(np.mean([ser_of_rho(b) for b in best]))
rows.append((Lp, K, float(best.mean()), ser_est))
print(f" L={Lp} done")
write_csv(DATA / "sec_brute.csv",
["L", "K", "best_rho", "eve_ser"], rows)
def main():
print(f"device={DEVICE}")
stage_A()
stage_B()
stage_C()
stage_D()
stage_E()
stage_F()
print("[done] full-scale security CSVs in", DATA)
if __name__ == "__main__":
main()
code/exp_kpa.py
+155
View File
@@ -0,0 +1,155 @@
"""Stage H: known-plaintext attack on the keyed masking.
The masking is linear in the keys, so an attacker who knows the indices
carried by some frames can write one linear equation per dimension per
frame and solve for the keys by least squares. This script measures how
much known plaintext the attacker needs before the recovered key is good
enough to decode, and how receiver noise slows that recovery down.
Per dimension k the observation of frame n is
y_k(n) = (1/c) sum_u e_{s_u(n),k} m_{u,k} + noise,
so stacking N frames gives A m_k = y_k with A(n,u) = e_{s_u(n),k}/c, an
N-by-U system that is solvable once N >= U in the noiseless case. The
attacker solves it per dimension, then correlates the estimate with the
true key and runs the correlation receiver with the estimated key.
Outputs:
kpa.csv : key correlation and eavesdropper SER against the number of
known-plaintext frames, at several SNRs
"""
from __future__ import annotations
import math
import numpy as np
import torch
from sse_lib import (DATA, DEVICE, SSE, rayleigh_gain, snr_to_sigma2,
set_seed, write_csv, eval_ser_sse)
from exp_full import get_model, eval_ser_eve
SNRS = [0.0, 10.0, 20.0]
NFRAMES = [1, 2, 3, 4, 5, 6, 8, 10, 12, 16, 24, 32, 48, 64]
SEED = 4242
EVAL_FRAMES = 50_000
# The spread across independent key-recovery attempts dominates the
# spread across channel realizations within one attempt, so the curve is
# smoothed by drawing many attempts rather than by lengthening each one.
TRIALS = 40
@torch.no_grad()
def collect_known_plaintext(model: SSE, n_frames: int, snr_db: float,
gen: torch.Generator):
"""Return (digits, raw observations, channel gains) for an attacker
that knows the transmitted indices.
The raw observation is returned rather than an equalized one. A
maximum-likelihood attacker keeps the channel gain in the design
matrix instead of dividing by it, which weights every frame by its
own quality and is the strongest use of the collected material. It
also avoids the numerical blow-up that equalizing a deep fade would
cause.
"""
digits = torch.randint(model.vu, (n_frames, model.users, model.P),
generator=gen).to(DEVICE)
Bn = model.unit_codebook()
m = model.masks()
e = Bn[digits] / math.sqrt(model.P)
y = (e * m[None, :, None, :]).sum(dim=1) / model.c # (N,P,L)
h = rayleigh_gain((n_frames,), device=DEVICE)
sigma = snr_to_sigma2(snr_db, model.d).to(DEVICE).sqrt()
noise = torch.randn(n_frames, model.P, model.L, device=DEVICE)
obs = h[:, None, None] * y + sigma * noise
return digits, obs, h
@torch.no_grad()
def solve_keys(model: SSE, digits, obs, h):
"""Maximum-likelihood key estimate from known plaintext.
Each period of each frame is an independent observation of the same
per-period key, so the P periods multiply the effective number of
equations. For entry l the system is A x = b with
A[(n,p), u] = h(n) e_{digit(n,u,p), l} / (c sqrt(P)) and b the raw
observation, so a frame in a deep fade contributes a small row on
both sides and is downweighted rather than amplified.
"""
Bn = model.unit_codebook() # (Vu, L)
N, U, P = digits.shape
L = model.L
c = float(model.c)
est = torch.zeros(U, L, device=DEVICE)
for l in range(L):
# design matrix over all (frame, period) pairs
A = Bn[digits, l] / (c * math.sqrt(P)) # (N,U,P)
A = A * h[:, None, None]
A = A.permute(0, 2, 1).reshape(N * P, U) # (N*P, U)
b = obs[:, :, l].reshape(N * P, 1) # (N*P, 1)
# A pseudo-inverse with an absolute tolerance is used instead of
# a least-squares driver. Training can leave a codebook entry
# numerically dead, with every codeword value below the smallest
# normal float. That entry carries no information about the
# digit, and inverting its system would amplify noise without
# bound, so the absolute tolerance discards it and the estimate
# for that entry stays at zero, which is what a careful attacker
# would do.
sol = torch.linalg.pinv(A.double(), atol=1e-12, rtol=0.0) @ b.double()
est[:, l] = sol[:, 0].float()
est = torch.nan_to_num(est)
# normalize to the key norm convention
est = est / est.norm(dim=1, keepdim=True).clamp_min(1e-9) * math.sqrt(L)
return est
def key_correlation(est: torch.Tensor, true: torch.Tensor) -> float:
"""Mean absolute normalized correlation over the users."""
a = est / est.norm(dim=1, keepdim=True).clamp_min(1e-9)
b = true / true.norm(dim=1, keepdim=True).clamp_min(1e-9)
return float((a * b).sum(dim=1).abs().mean())
def main():
set_seed(SEED)
model = get_model(iters=4000)
model.eval()
true_m = model.masks().detach()
legit = eval_ser_sse(model, [10.0], frames=200_000)[0]
print(f"[kpa] legitimate SER at 10 dB = {legit:.4g}, U={model.users}, "
f"L={model.L}")
# Nested known-plaintext sets with common random numbers. Within a
# trial the attacker collects one pool of frames and the N-frame
# estimate uses the first N of them, so more material can only help,
# exactly as an attacker accumulating traffic would experience. The
# evaluation noise is also shared across N within a trial. Both
# choices remove the between-point variance that would otherwise make
# the averaged curve jagged, without changing what is being measured.
nmax = max(NFRAMES)
rows = []
for snr in SNRS:
acc = {n: [[], []] for n in NFRAMES}
for t in range(TRIALS):
gen = torch.Generator(device="cpu").manual_seed(
SEED + int(snr) + 1000 * t)
digits, obs, h = collect_known_plaintext(model, nmax, snr, gen)
eval_seed = 777 + 31 * t + int(snr)
for n in NFRAMES:
est = solve_keys(model, digits[:n], obs[:n], h[:n])
acc[n][0].append(key_correlation(est, true_m))
acc[n][1].append(eval_ser_eve(model, est.cpu(), [10.0],
frames=EVAL_FRAMES,
seed=eval_seed)[0])
for n in NFRAMES:
ks, ss = acc[n]
kappa = sum(ks) / len(ks)
ser = sum(ss) / len(ss)
rows.append((snr, n, kappa, ser))
print(f" snr={snr:4.1f} N={n:5d} kappa={kappa:.4f} "
f"eve_ser={ser:.4f}")
write_csv(DATA / "kpa.csv",
["snr_db", "n_frames", "kappa", "eve_ser"], rows)
print("[done] kpa.csv")
if __name__ == "__main__":
main()
code/exp_real_sec.py
+195
View File
@@ -0,0 +1,195 @@
"""Stage G: security on real language-model token streams.
AG News test headlines are tokenized with the bert-base-uncased
WordPiece tokenizer (vocabulary 30,522). Four users carry four disjoint
headline streams, each frame transmits one token per user, and the token
identifier is carried by its base-16 digits, so the digit space
16^4 = 65,536 covers the vocabulary. The keys and codebook trained on
uniform indices are reused unchanged, so this stage tests the design on
a real, highly non-uniform source without retraining.
Two metrics are reported. The token error rate is the symbol-level
measure used in the rest of the paper. The headline recovery rate is a
meaning-level measure: the fraction of complete headlines a receiver
reconstructs without a single token error, which is what an
eavesdropper actually needs to read the message.
Outputs:
real_sec_ter.csv : token error rate vs SNR for legitimate, outsider
eavesdropper, insider, and OMA
real_sec_stats.json: stream statistics and headline recovery rates
"""
from __future__ import annotations
import json
import math
import torch
import sse_lib as L
from sse_lib import (DATA, DEVICE, SSE, rayleigh_gain, snr_to_sigma2,
set_seed, write_csv)
from exp_full import get_model, eve_wrong_mask
SNR_GRID = [0, 4, 8, 12, 16, 20, 24, 28]
# headline recovery is meaningful only where the legitimate user clears
# most tokens, since a headline averages tens of tokens and needs every
# one of them correct
REC_SNR = (20, 24, 28)
N_TEXTS = 2000
REPEATS = 8
REC_RUNS = 4
SEED_EVAL = 777
P_MAX, VU, U = 4, 16, 4
def load_streams():
from datasets import load_dataset
from transformers import AutoTokenizer
tok = AutoTokenizer.from_pretrained("bert-base-uncased")
ds = load_dataset("fancyzhx/ag_news", split="test")
texts = [ds[i]["text"] for i in range(N_TEXTS)]
streams = [[] for _ in range(U)]
bounds = [[] for _ in range(U)] # (start, end) per headline
for i, t in enumerate(texts):
ids = tok(t, add_special_tokens=False)["input_ids"]
u = i % U
s = len(streams[u])
streams[u].extend(ids)
bounds[u].append((s, s + len(ids)))
n = min(len(s) for s in streams)
streams = [s[:n] for s in streams]
bounds = [[(a, b) for (a, b) in bu if b <= n] for bu in bounds]
return streams, bounds, tok.vocab_size
def ids_to_digits(ids: torch.Tensor) -> torch.Tensor:
"""(N,U) token ids -> (N,U,P) base-16 digits, most significant first."""
d, x = [], ids.clone()
for _ in range(P_MAX):
d.append(x % VU)
x = x // VU
return torch.stack(d[::-1], dim=-1)
@torch.no_grad()
def wrong_keyed(model: SSE, digits_all, snr_db, seed, rx_masks=None,
chunk=50_000):
"""Per-frame per-user error indicator (N,U) for a receiver that
correlates with rx_masks. rx_masks=None means the legitimate keys."""
torch.manual_seed(seed)
Bn = model.unit_codebook()
true_m = model.masks()
rx = true_m if rx_masks is None else rx_masks.to(DEVICE)
c = model.c
N = digits_all.shape[0]
wrong = torch.zeros(N, U, dtype=torch.bool)
sigma = snr_to_sigma2(snr_db, model.d).to(DEVICE).sqrt()
for n0 in range(0, N, chunk):
dg = digits_all[n0:n0 + chunk].to(DEVICE)
n = dg.shape[0]
e = Bn[dg] / math.sqrt(model.P)
y = (e * true_m[None, :, None, :]).sum(dim=1) / c
h = rayleigh_gain((n, U), device=DEVICE)
noise = torch.randn(n, U, model.P, model.L, device=DEVICE)
y_rx = h[:, :, None, None] * y[:, None] + sigma * noise
r = y_rx / h[:, :, None, None].clamp_min(1e-6)
cand = Bn[None, :, :] * rx[:, None, :]
sc = torch.einsum("nupl,uvl->nupv", r, cand)
wrong[n0:n0 + chunk] = (sc.argmax(-1) != dg).any(dim=2).cpu()
return wrong
@torch.no_grad()
def wrong_oma(ids_all, snr_db, seed, bits=16):
"""Antipodal signaling on the actual token bits, same frame energy."""
torch.manual_seed(seed)
N, Uu = ids_all.shape
b = ((ids_all[..., None] >> torch.arange(bits)) & 1).float() * 2 - 1
b = b.to(DEVICE)
sigma = math.sqrt(1.0 / (10.0 ** (snr_db / 10.0)))
h = rayleigh_gain((N, Uu, 1))
y = h * b + sigma * torch.randn(N, Uu, bits, device=DEVICE)
return ((y * b) < 0).any(dim=2).cpu()
def headline_recovery(wrong: torch.Tensor, bounds) -> tuple[int, int]:
"""A headline counts as recovered only if every token is correct."""
ok = tot = 0
for u in range(U):
wu = wrong[:, u]
for (a, b) in bounds[u]:
tot += 1
ok += int(not bool(wu[a:b].any()))
return ok, tot
def main():
set_seed(SEED_EVAL)
streams, bounds, vocab = load_streams()
ids_all = torch.tensor(list(zip(*streams)), dtype=torch.long) # (N,U)
digits_all = ids_to_digits(ids_all)
N = ids_all.shape[0]
assert int(ids_all.max()) < VU ** P_MAX
print(f"[real] {N} frames, {int(torch.unique(ids_all).numel())} "
f"distinct tokens, max id {int(ids_all.max())}")
# keys and codebook trained on uniform indices, reused unchanged
model = get_model(P=P_MAX, vu=VU, d=64, U=U, iters=4000)
model.eval()
eve_m = eve_wrong_mask(U, model.L, seed=20260813) # outsider
ins_m = model.masks().detach().roll(1, 0).cpu() # insider
schemes = {
"legit": lambda s, k: wrong_keyed(model, digits_all, s, k),
"eve": lambda s, k: wrong_keyed(model, digits_all, s, k, eve_m),
"insider": lambda s, k: wrong_keyed(model, digits_all, s, k, ins_m),
"oma": lambda s, k: wrong_oma(ids_all, s, k),
}
rows = []
for s in SNR_GRID:
ter = {}
for name, fn in schemes.items():
e = 0
for r in range(REPEATS):
e += int(fn(s, SEED_EVAL + 1000 * r + int(10 * s)).sum())
ter[name] = e / (N * U * REPEATS)
rows.append((s, ter["legit"], ter["eve"], ter["insider"], ter["oma"]))
print("[real]", [f"{v:.4g}" for v in rows[-1]])
write_csv(DATA / "real_sec_ter.csv",
["snr_db", "ter_legit", "ter_eve", "ter_insider", "ter_oma"],
rows)
rec = {}
for s in REC_SNR:
rec[str(s)] = {}
for name, fn in schemes.items():
ok = tot = 0
for r in range(REC_RUNS):
w = fn(s, SEED_EVAL + 5000 * r + int(10 * s))
o, t = headline_recovery(w, bounds)
ok += o; tot += t
rec[str(s)][name] = ok / tot
print(f"[rec] {s} dB {name}: {ok}/{tot} = {ok/tot:.4g}")
stats = {
"vocab_size": vocab,
"n_texts": N_TEXTS,
"frames": N,
"repeats": REPEATS,
"decisions_per_point": N * U * REPEATS,
"distinct_tokens": int(torch.unique(ids_all).numel()),
"max_token_id": int(ids_all.max()),
"headlines_scored": sum(len(b) for b in bounds),
"headline_runs": REC_RUNS,
"recovery": rec,
}
(DATA / "real_sec_stats.json").write_text(json.dumps(stats, indent=1))
print(json.dumps(stats, indent=1))
if __name__ == "__main__":
main()
code/feasibility_security.py
+208
View File
@@ -0,0 +1,208 @@
"""Feasibility study for paper 11 (TIFS): the per-user mask as a
physical-layer key.
Three questions, all under the shared-embedding multiple-access model of
sse_lib.py (real-vector convention, flat Rayleigh fading):
Q1 (encryption): a legitimate receiver knows its mask mu_u; an
eavesdropper (Eve) does not. How far above chance can Eve decode?
We measure the legitimate symbol error rate (SER) against Eve's SER
when Eve applies (a) a wrong mask drawn from the same distribution,
(b) no mask (mu = 1), (c) the average mask. Chance level is
(Vu-1)/Vu per digit, 1-(1/Vu)^P per frame.
Q2 (key entropy vs dimension): as the per-period length L grows, two
independently drawn unit-norm masks become more nearly orthogonal,
so Eve's residual after de-masking with a wrong key grows. We sweep
L and report Eve's SER and the mean absolute mask cross-correlation.
Q3 (jamming robustness): a jammer adds h_J * w to the frame, where w is
an arbitrary unit waveform (worst case: aligned with the victim's
masked codeword direction; and random). We sweep the
jammer-to-signal ratio (JSR) and report the legitimate SER, to show
the mask spreads a mismatched jammer and bounds its effect.
This is a CPU-sized feasibility run (small V), not the final experiment.
Seeds fixed; results written to ../data as CSV.
"""
from __future__ import annotations
import math
import numpy as np
import torch
import sse_lib as L
from sse_lib import SSE, rayleigh_gain, snr_to_sigma2, write_csv, set_seed, DATA, DEVICE
# ----------------------------------------------------------------------
# Eve: apply a chosen (wrong) set of masks to the SAME received frame the
# legitimate users see, then run the correlation receiver.
# ----------------------------------------------------------------------
@torch.no_grad()
def eval_ser_eve(model: SSE, eve_masks: torch.Tensor, snr_list,
frames: int = 400_000, chunk: int = 50_000, seed: int = 777):
"""eve_masks: (U, L) the masks Eve uses in place of the true ones.
Eve observes the same physically transmitted frame (true masks used at
the transmitter) but correlates with eve_masks."""
model.eval().to(DEVICE)
Bn = model.unit_codebook()
true_m = model.masks()
eve_masks = eve_masks.to(DEVICE)
c = model.c
out = []
for snr_db in snr_list:
g = torch.Generator(device="cpu").manual_seed(seed + int(10 * snr_db))
err = tot = 0
for n0 in range(0, frames, chunk):
n = min(chunk, frames - n0)
digits = torch.randint(model.vu, (n, model.users, model.P),
generator=g).to(DEVICE)
# transmit with the TRUE masks
e = Bn[digits] / math.sqrt(model.P)
y = (e * true_m[None, :, None, :]).sum(dim=1) / c # (n,P,L)
h = rayleigh_gain((n, model.users), device=DEVICE)
sigma = snr_to_sigma2(snr_db).to(DEVICE).sqrt()
noise = torch.randn(n, model.users, model.P, model.L, device=DEVICE)
y_rx = h[:, :, None, None] * y[:, None] + sigma * noise
r = y_rx / h[:, :, None, None].clamp_min(1e-6) # (n,U,P,L)
# Eve correlates with her (wrong) masks
cand = Bn[None, :, :] * eve_masks[:, None, :] # (U,Vu,L)
scores = torch.einsum("nupl,uvl->nupv", r, cand)
wrong = (scores.argmax(-1) != digits).any(dim=2)
err += int(wrong.sum()); tot += n * model.users
out.append(err / tot)
return out
@torch.no_grad()
def eval_ser_jam(model: SSE, snr_db, jsr_db_list, frames: int = 400_000,
chunk: int = 50_000, seed: int = 777, mode: str = "aligned"):
"""Legitimate SER with an added jammer h_J * sqrt(JSR) * w.
mode='aligned': w points along user 0's masked mean codeword direction
(a structured, mask-matched worst case for user 0).
mode='random': w is an isotropic random unit frame each transmission."""
model.eval().to(DEVICE)
Bn = model.unit_codebook()
true_m = model.masks()
c = model.c
sigma = snr_to_sigma2(snr_db).to(DEVICE).sqrt()
# aligned jammer direction: mask-0 applied to a fixed unit codeword,
# i.e. what an attacker would build if it copied the public codebook
# but guessed the (secret) mask wrong -> here we give it mask 0 exactly
# as the strongest realistic structured jammer.
w_fixed = (Bn[0][None, :] * true_m[0][None, :]).repeat(model.P, 1) # (P,L)
w_fixed = w_fixed / w_fixed.norm()
out = []
for jsr_db in jsr_db_list:
jsr = 10.0 ** (jsr_db / 10.0)
g = torch.Generator(device="cpu").manual_seed(seed + int(10 * jsr_db))
err = tot = 0
for n0 in range(0, frames, chunk):
n = min(chunk, frames - n0)
digits = torch.randint(model.vu, (n, model.users, model.P),
generator=g).to(DEVICE)
e = Bn[digits] / math.sqrt(model.P)
y = (e * true_m[None, :, None, :]).sum(dim=1) / c # (n,P,L)
h = rayleigh_gain((n, model.users), device=DEVICE)
hJ = rayleigh_gain((n,), device=DEVICE)
if mode == "aligned":
w = w_fixed[None].expand(n, model.P, model.L)
else:
w = torch.randn(n, model.P, model.L, device=DEVICE)
w = w / w.reshape(n, -1).norm(dim=1)[:, None, None].clamp_min(1e-8)
jam = (hJ * math.sqrt(jsr))[:, None, None] * w # (n,P,L)
noise = torch.randn(n, model.users, model.P, model.L, device=DEVICE)
y_rx = (h[:, :, None, None] * y[:, None]
+ h[:, :, None, None] * 0 # keep shape clarity
+ jam[:, None] + sigma * noise)
r = y_rx / h[:, :, None, None].clamp_min(1e-6)
cand = Bn[None, :, :] * true_m[:, None, :]
scores = torch.einsum("nupl,uvl->nupv", r, cand)
wrong = (scores.argmax(-1) != digits).any(dim=2)
err += int(wrong.sum()); tot += n * model.users
out.append(err / tot)
return out
def mean_abs_cross_corr(masks: torch.Tensor) -> float:
"""Mean |<mu_i, mu_j>| / (||mu_i|| ||mu_j||) over i<j."""
m = masks / masks.norm(dim=1, keepdim=True).clamp_min(1e-8)
G = (m @ m.T).abs()
U = m.shape[0]
off = G[~torch.eye(U, dtype=torch.bool, device=G.device)]
return float(off.mean())
def main():
set_seed(1)
# CPU-sized feasibility configuration: V = Vu^P = 16^2 = 256
P, VU, D, U = 2, 16, 64, 4
snr_eval = [0.0, 5.0, 10.0, 15.0, 20.0]
chance_frame = 1.0 - (1.0 / VU) ** P
model = SSE(P=P, vu=VU, d=D, users=U).to(DEVICE)
print(f"[train] SSE P={P} Vu={VU} d={D} U={U} V={model.V} on {DEVICE}")
L.TRAIN_SNR_DB = (0.0, 20.0)
model_iters = 1500
curve = L.train_sse(model, iters=model_iters, batch=256, lr=3e-3,
log_every=0, seed=1)
model.calibrate_power()
legit = L.eval_ser_sse(model, snr_eval, frames=400_000)
print("[Q1] legitimate SER:", [f"{v:.3g}" for v in legit])
# Eve variants
set_seed(20260813)
eve_wrong = torch.randn(U, model.L) / math.sqrt(model.L)
eve_wrong = eve_wrong / eve_wrong.norm(dim=1, keepdim=True) * math.sqrt(model.L)
eve_none = torch.ones(U, model.L)
eve_avg = model.masks().mean(dim=0, keepdim=True).repeat(U, 1).cpu()
eve_w = eval_ser_eve(model, eve_wrong, snr_eval, frames=400_000)
eve_n = eval_ser_eve(model, eve_none, snr_eval, frames=400_000)
eve_a = eval_ser_eve(model, eve_avg, snr_eval, frames=400_000)
print("[Q1] Eve wrong-mask SER:", [f"{v:.3g}" for v in eve_w])
print("[Q1] Eve no-mask SER:", [f"{v:.3g}" for v in eve_n])
print(f"[Q1] chance frame SER = {chance_frame:.4f}")
write_csv(DATA / "feas_q1_eavesdrop.csv",
["snr_db", "legit", "eve_wrong", "eve_none", "eve_avg", "chance"],
[(s, legit[i], eve_w[i], eve_n[i], eve_a[i], chance_frame)
for i, s in enumerate(snr_eval)])
# Q2: key entropy vs per-period length L (grow d at fixed P)
print("[Q2] sweeping period length L ...")
q2_rows = []
for d in [16, 32, 64, 128, 256]:
set_seed(1)
mdl = SSE(P=P, vu=VU, d=d, users=U).to(DEVICE)
L.train_sse(mdl, iters=model_iters, batch=256, lr=3e-3, seed=1)
mdl.calibrate_power()
set_seed(20260813)
ew = torch.randn(U, mdl.L) / math.sqrt(mdl.L)
ew = ew / ew.norm(dim=1, keepdim=True) * math.sqrt(mdl.L)
lg = L.eval_ser_sse(mdl, [10.0], frames=300_000)[0]
ev = eval_ser_eve(mdl, ew, [10.0], frames=300_000)[0]
xc = mean_abs_cross_corr(mdl.masks().detach().cpu())
q2_rows.append((mdl.L, d, lg, ev, xc))
print(f" L={mdl.L:4d} legit={lg:.3g} eve={ev:.3g} |xcorr|={xc:.3f}")
write_csv(DATA / "feas_q2_keyentropy.csv",
["L", "d", "legit_ser", "eve_ser", "mask_xcorr"], q2_rows)
# Q3: jamming robustness at SNR=10 dB
print("[Q3] jamming sweep at SNR=10 dB ...")
jsr = [-10.0, -5.0, 0.0, 5.0, 10.0, 15.0, 20.0]
jam_al = eval_ser_jam(model, 10.0, jsr, frames=300_000, mode="aligned")
jam_rd = eval_ser_jam(model, 10.0, jsr, frames=300_000, mode="random")
print("[Q3] aligned-jammer SER:", [f"{v:.3g}" for v in jam_al])
print("[Q3] random-jammer SER:", [f"{v:.3g}" for v in jam_rd])
write_csv(DATA / "feas_q3_jamming.csv",
["jsr_db", "ser_aligned", "ser_random"],
[(j, jam_al[i], jam_rd[i]) for i, j in enumerate(jsr)])
print("\n[done] feasibility CSVs written to", DATA)
if __name__ == "__main__":
main()
code/make_tables.py
+55
View File
@@ -0,0 +1,55 @@
"""Generate the LaTeX rows of the two result tables from the CSVs, so
that every table in the paper is reproducible from data/ (TIFS mandate).
Prints the tabular body; paste into main.tex without edits.
"""
from __future__ import annotations
import csv
from pathlib import Path
DATA = Path(__file__).resolve().parents[1] / "data"
NAME = {
"proposed": r"\textbf{Proposed keyed masking}",
"public_mask": "Public masks",
"perm_key": r"Global-permutation key~\cite{chen2023shuffling}",
"index_cipher": "Per-user index cipher",
"oma_plain": "OMA (no encryption)",
"random": "Random",
"hadamard": "Walsh-Hadamard",
"learned": "Learned",
}
def f3(x: str) -> str:
try:
return f"{float(x):.3f}"
except ValueError:
return "--"
def compare_table():
print("% Table: scheme comparison (from sec_compare.csv)")
rows = list(csv.DictReader(open(DATA / "sec_compare.csv")))
order = ["public_mask", "perm_key", "index_cipher", "oma_plain", "proposed"]
rows = sorted(rows, key=lambda r: order.index(r["scheme"]))
for r in rows:
cells = [f3(r["legit_ser"]), f3(r["eve_out"]), f3(r["eve_in"]),
f3(r["jam0_ser"])]
if r["scheme"] == "proposed":
cells = [rf"$\mathbf{{{c}}}$" for c in cells]
else:
cells = [f"${c}$" if c != "--" else "--" for c in cells]
print(f"{NAME[r['scheme']]} & " + " & ".join(cells) + r" \\")
def maskfam_table():
print("% Table: key families (from sec_maskfam.csv)")
for r in csv.DictReader(open(DATA / "sec_maskfam.csv")):
print(f"{NAME[r['family']]} & ${f3(r['legit_ser'])}$ & "
f"${f3(r['eve_ser'])}$ & ${f3(r['mask_xcorr'])}$" + r" \\")
if __name__ == "__main__":
compare_table()
print()
maskfam_table()
code/replot_security.py
+253
View File
@@ -0,0 +1,253 @@
"""Canonical replot script for paper 11: regenerates every result figure
from ../data/*.csv and writes paper-ready PDFs to ../fig/. No experiment
is rerun. All result plots share one canvas and axes rectangle (8:6 box).
Label dictionary is fixed here and copied verbatim into tables and prose.
fig_sec_snr.pdf : legitimate vs eavesdropper SER vs SNR (Fig. 2)
fig_sec_keylen.pdf : SER vs key length L (Fig. 3)
fig_sec_jam.pdf : target-user SER vs JSR (Fig. 4)
fig_sec_sens.pdf : Eve SER vs key correlation (Fig. 5)
fig_sec_brute.pdf : Eve SER vs number of key guesses (Fig. 6)
"""
from __future__ import annotations
from pathlib import Path
import csv
import matplotlib
matplotlib.use("Agg")
import matplotlib.pyplot as plt
ROOT = Path(__file__).resolve().parents[1]
DATA = ROOT / "data"
FIG = ROOT / "fig"
FIG.mkdir(exist_ok=True)
plt.rcParams.update({
"font.family": "serif",
"font.serif": ["DejaVu Serif", "Times New Roman"],
"font.size": 9,
"axes.labelsize": 9,
"legend.fontsize": 6.6,
"xtick.labelsize": 8,
"ytick.labelsize": 8,
"axes.grid": True,
"grid.linestyle": "--",
"grid.linewidth": 0.4,
"grid.alpha": 0.6,
"lines.linewidth": 1.3,
"lines.markersize": 3.4,
"figure.figsize": (3.15, 2.36),
"pdf.fonttype": 42,
})
AXES_RECT = dict(left=0.185, right=0.965, top=0.955, bottom=0.195)
C_LEGIT = "#c0392b"
C_EVE = "#2c5fa8"
C_OMA = "#7f8c8d"
C_CH = "#95a5a6"
C_MATCH = "#8e44ad"
C_PUB = "#16a085"
# fixed label dictionary: tables and prose copy these strings verbatim
LBL = {
"legit": "Legitimate",
"oma": "OMA",
"eve_pub": "Eve, public masks",
"eve_key": "Eve, wrong key",
"chance": "Random guess",
"jam_m": "Matched jammer (public masks)",
"jam_b": "Blind jammer (proposed)",
"nojam": "No jammer",
}
def load(name):
with open(DATA / name) as f:
return list(csv.DictReader(f))
def col(rows, k, f=float):
return [f(r[k]) for r in rows]
def save(fig, name):
"""Write the figure and assert that no axis label is clipped.
A long y label, or wide minor tick labels such as 6x10^-1 on a log
axis that spans less than a decade, silently pushes the label off
the canvas under the fixed axes rectangle. Reading the plotting code
cannot reveal this, so the check is made on the rendered geometry.
"""
fig.subplots_adjust(**AXES_RECT)
fig.canvas.draw()
fbox = fig.get_window_extent()
for ax in fig.axes:
for lbl in (ax.yaxis.label, ax.xaxis.label):
if not lbl.get_text():
continue
b = lbl.get_window_extent()
if (b.x0 < fbox.x0 or b.y0 < fbox.y0
or b.x1 > fbox.x1 or b.y1 > fbox.y1):
raise RuntimeError(
f"{name}: axis label '{lbl.get_text()}' is clipped "
f"(label {b} outside figure {fbox}); shorten the "
f"label or widen the margin")
fig.savefig(FIG / f"{name}.pdf")
plt.close(fig)
print("[OK]", name)
def fig_snr():
r = load("sec_snr.csv")
x = col(r, "snr_db")
fig, ax = plt.subplots()
ax.semilogy(x, col(r, "legit"), color=C_LEGIT, marker="o", ls="-",
label=LBL["legit"])
ax.semilogy(x, col(r, "oma"), color=C_OMA, marker="^", ls=":",
label=LBL["oma"])
ax.semilogy(x, col(r, "eve_public"), color=C_PUB, marker="v",
ls="none", markersize=5.2, markerfacecolor="none",
label=LBL["eve_pub"])
ax.semilogy(x, col(r, "eve_wrong"), color=C_EVE, marker="s", ls="--",
label=LBL["eve_key"])
ax.plot(x, col(r, "chance"), color=C_CH, ls="-.", lw=0.9,
label=LBL["chance"])
ax.set_xlabel("SNR (dB)")
ax.set_ylabel("SER")
ax.set_xlim(min(x), max(x))
ax.legend(loc="lower left")
save(fig, "fig_sec_snr")
def fig_keylen():
r = load("sec_keylen.csv")
x = col(r, "L", int)
fig, ax = plt.subplots()
ax.semilogy(x, col(r, "legit_ser"), color=C_LEGIT, marker="o", ls="-",
label=LBL["legit"])
ax.semilogy(x, col(r, "oma"), color=C_OMA, marker="^", ls=":",
label=LBL["oma"])
ax.semilogy(x, col(r, "eve_ser"), color=C_EVE, marker="s", ls="--",
label=LBL["eve_key"])
ax.set_xlabel("Key length $L$")
ax.set_ylabel("SER")
ax.set_xscale("log", base=2)
ax.legend(loc="center right", bbox_to_anchor=(0.98, 0.72))
save(fig, "fig_sec_keylen")
def fig_jam():
# the target-user SER spans 0.3 to 1.0, less than one decade, so a
# linear axis is used: a log axis here produces wide minor tick
# labels (6x10^-1) that crowd out the y label under the fixed
# axes rectangle
r = load("sec_jam.csv")
x = col(r, "jsr_db")
fig, ax = plt.subplots()
ax.plot(x, col(r, "matched"), color=C_MATCH, marker="P", ls="--",
label=LBL["jam_m"])
ax.plot(x, col(r, "blind"), color=C_LEGIT, marker="o", ls="-",
label=LBL["jam_b"])
nojam = col(r, "nojam")[0]
ax.axhline(nojam, color=C_OMA, ls=":", lw=0.9, label=LBL["nojam"])
ax.set_xlabel("JSR (dB)")
ax.set_ylabel("SER")
ax.set_xlim(min(x), max(x))
ax.set_ylim(0.2, 1.02)
ax.legend(loc="lower right")
save(fig, "fig_sec_jam")
def fig_sens():
r = load("sec_sens.csv")
x = col(r, "rho")
fig, ax = plt.subplots()
ax.plot(x, col(r, "eve_ser"), color=C_EVE, marker="s", ls="-",
label=LBL["eve_key"])
chance = 1.0 - (1.0 / 16.0) ** 4
ax.axhline(chance, color=C_CH, ls="-.", lw=0.9, label=LBL["chance"])
ax.set_xlabel(r"Key correlation $\kappa$")
ax.set_ylabel("Eavesdropper SER")
ax.set_xlim(0, 1)
ax.legend(loc="lower left")
save(fig, "fig_sec_sens")
def fig_brute():
r = load("sec_brute.csv")
fig, ax = plt.subplots()
sty = {8: ("#c0392b", "o"), 16: ("#2c5fa8", "s"),
32: ("#16a085", "v"), 64: ("#8e44ad", "P")}
for Lp in [8, 16, 32, 64]:
rows = [row for row in r if int(row["L"]) == Lp]
ks = [float(row["K"]) for row in rows]
ser = [float(row["eve_ser"]) for row in rows]
c, mk = sty[Lp]
ax.semilogx(ks, ser, color=c, marker=mk, ls="-",
label=f"$L={Lp}$")
ax.set_xlabel("Number of key guesses $K$")
ax.set_ylabel("Eavesdropper SER")
ax.legend(loc="lower left")
save(fig, "fig_sec_brute")
def fig_real():
r = load("real_sec_ter.csv")
x = col(r, "snr_db")
fig, ax = plt.subplots()
ax.semilogy(x, col(r, "ter_legit"), color=C_LEGIT, marker="o", ls="-",
label=LBL["legit"])
ax.semilogy(x, col(r, "ter_oma"), color=C_OMA, marker="^", ls=":",
label=LBL["oma"])
ax.semilogy(x, col(r, "ter_insider"), color=C_PUB, marker="v", ls="-.",
label="Insider")
ax.semilogy(x, col(r, "ter_eve"), color=C_EVE, marker="s", ls="--",
label=LBL["eve_key"])
ax.set_xlabel("SNR (dB)")
ax.set_ylabel("TER")
ax.set_xlim(min(x), max(x))
ax.legend(loc="lower left")
save(fig, "fig_sec_real")
def fig_kpa():
r = load("kpa.csv")
fig, ax = plt.subplots()
sty = {0.0: ("#c0392b", "o"), 10.0: ("#2c5fa8", "s"),
20.0: ("#16a085", "v")}
for snr, (c, mk) in sty.items():
rows = [row for row in r if float(row["snr_db"]) == snr]
n = [float(row["n_frames"]) for row in rows]
ser = [float(row["eve_ser"]) for row in rows]
ax.semilogx(n, ser, color=c, marker=mk, ls="-",
label=f"{int(snr)} dB")
ax.axhline(0.304, color=C_OMA, ls=":", lw=0.9, label=LBL["legit"])
ax.set_xlabel("Known-plaintext frames $N$")
ax.set_ylabel("Eavesdropper SER")
ax.set_xscale("log", base=2)
ax.legend(loc="upper right")
save(fig, "fig_sec_kpa")
def main():
fig_snr()
fig_keylen()
fig_jam()
try:
fig_sens()
fig_brute()
except FileNotFoundError:
print("[skip] attack-difficulty CSVs not present yet")
try:
fig_real()
except FileNotFoundError:
print("[skip] real-token CSV not present yet")
try:
fig_kpa()
except FileNotFoundError:
print("[skip] known-plaintext CSV not present yet")
print("[done] figures in", FIG)
if __name__ == "__main__":
main()
code/sse_lib.py
+388
View File
@@ -0,0 +1,388 @@
"""Shared library for the scalable shared embedding (SSE) letter.
System model (real-vector convention, declared in the paper):
d-dimensional real embedding frame, U users, single transmitter.
Proposed SSE: the frame is split into P periods of length L = d/P and
one unit codebook B in R^{Vu x L} is reused in every period, so the
vocabulary size is V = Vu^P while the codebook stores only Vu*L
numbers. Index v maps to base-Vu digits (i_1,...,i_P).
Per-user periodic masks mu_u in R^L are repeated over the P periods.
Tx: y = (1/c) * sum_u e(s_u) .* m_u, c fixes unit average frame power.
Channel: user u sees h_u * y + n, h_u^2 ~ Exp(1) (Rayleigh magnitude,
known at the receiver), n ~ N(0, sigma^2 I), sigma^2 = 1/(d*snr).
Rx u: equalize by h_u, per period correlate with the masked unit
codewords b_i .* mu_u and take the argmax digit.
Device: CUDA when available (run under WSL), CPU fallback.
"""
from __future__ import annotations
import math
import time
from pathlib import Path
import numpy as np
import torch
ROOT = Path(__file__).resolve().parents[1]
DATA = ROOT / "data"
FIG = ROOT / "fig"
DATA.mkdir(exist_ok=True)
FIG.mkdir(exist_ok=True)
DEVICE = torch.device("cuda" if torch.cuda.is_available() else "cpu")
# ----------------------------------------------------------------------
# global configuration
# ----------------------------------------------------------------------
D = 64 # embedding dimension (real)
U = 4 # users
VU = 16 # unit codebook size
P_MAX = 4 # periods for the main configuration, V = 16^4 = 65536
SEED = 1
TRAIN_ITERS = 4000
TRAIN_BATCH = 256
TRAIN_SNR_DB = (0.0, 20.0)
LR = 3e-3
def set_seed(seed: int = SEED) -> None:
torch.manual_seed(seed)
np.random.seed(seed)
if torch.cuda.is_available():
torch.cuda.manual_seed_all(seed)
def snr_to_sigma2(snr_db: torch.Tensor | float,
d: int = D) -> torch.Tensor | float:
"""Per-dimension noise variance for unit frame power and E[h^2]=1.
Pass the model's actual frame dimension d when it differs from the
module default, otherwise the effective SNR shifts with d."""
snr = 10.0 ** (torch.as_tensor(snr_db, dtype=torch.float64) / 10.0)
return (1.0 / (d * snr)).float()
def rayleigh_gain(shape, device=DEVICE) -> torch.Tensor:
"""|g| with g ~ CN(0,1): h^2 ~ Exp(1), E[h^2] = 1."""
u = torch.rand(shape, device=device).clamp_min(1e-12)
return torch.sqrt(-torch.log(u))
# ----------------------------------------------------------------------
# proposed scalable shared embedding
# ----------------------------------------------------------------------
class SSE(torch.nn.Module):
"""Periodic unit-codebook shared embedding with per-user periodic masks."""
def __init__(self, P: int = P_MAX, vu: int = VU, d: int = D, users: int = U):
super().__init__()
assert d % P == 0, "embedding dimension must split into P periods"
self.P, self.vu, self.d, self.users = P, vu, d, users
self.L = d // P
self.B = torch.nn.Parameter(torch.randn(vu, self.L) / math.sqrt(self.L))
self.W = torch.nn.Parameter(torch.randn(users, self.L) / math.sqrt(self.L))
self.logit_scale = torch.nn.Parameter(torch.tensor(2.0))
# transmit power normalizer, calibrated after training (buffer)
self.register_buffer("c", torch.tensor(1.0))
@property
def V(self) -> int:
return self.vu ** self.P
def unit_codebook(self) -> torch.Tensor:
return self.B / self.B.norm(dim=1, keepdim=True).clamp_min(1e-8)
def masks(self) -> torch.Tensor:
return self.W / self.W.norm(dim=1, keepdim=True).clamp_min(1e-8) * math.sqrt(self.L)
def tx_frame(self, digits: torch.Tensor) -> torch.Tensor:
"""digits: (N, U, P) ints -> unnormalized tx frame (N, P, L)."""
Bn = self.unit_codebook()
e = Bn[digits] / math.sqrt(self.P) # (N,U,P,L)
m = self.masks() # (U,L)
x = e * m[None, :, None, :]
return x.sum(dim=1) # (N,P,L)
def calibrate_power(self, n: int = 65536) -> None:
with torch.no_grad():
digits = torch.randint(self.vu, (n, self.users, self.P), device=self.B.device)
y = self.tx_frame(digits)
self.c.fill_(float(y.pow(2).sum(dim=(1, 2)).mean().sqrt()))
def scores(self, r: torch.Tensor) -> torch.Tensor:
"""r: equalized rx frame (N,P,L) -> scores (N,U,P,Vu)."""
Bn = self.unit_codebook() # (Vu,L)
m = self.masks() # (U,L)
cand = Bn[None, :, :] * m[:, None, :] # (U,Vu,L)
return torch.einsum("npl,uvl->nupv", r, cand)
def forward(self, digits: torch.Tensor, snr_db: torch.Tensor,
h: torch.Tensor | None = None):
"""digits (N,U,P), snr_db (N,) -> per-user scores and rx frames."""
N = digits.shape[0]
y = self.tx_frame(digits) / self.c # (N,P,L)
if h is None:
h = rayleigh_gain((N, self.users), device=y.device)
sigma = snr_to_sigma2(snr_db, self.d).to(y.device).sqrt() # (N,)
n = torch.randn(N, self.users, self.P, self.L, device=y.device)
y_rx = h[:, :, None, None] * y[:, None] + sigma[:, None, None, None] * n
r = y_rx / h[:, :, None, None].clamp_min(1e-6) # equalized (N,U,P,L)
Bn = self.unit_codebook()
m = self.masks()
cand = Bn[None, :, :] * m[:, None, :] # (U,Vu,L)
scores = torch.einsum("nupl,uvl->nupv", r, cand)
return scores
def n_params(self) -> int:
return self.B.numel() + self.W.numel()
# ----------------------------------------------------------------------
# conventional scheme: full unstructured codebook (prior shared embedding)
# ----------------------------------------------------------------------
class FullCodebook(torch.nn.Module):
"""Directly trained V x d codebook with full-d per-user masks."""
def __init__(self, V: int, d: int = D, users: int = U):
super().__init__()
self.Vc, self.d, self.users = V, d, users
self.E = torch.nn.Parameter(torch.randn(V, d) / math.sqrt(d))
self.W = torch.nn.Parameter(torch.randn(users, d) / math.sqrt(d))
self.logit_scale = torch.nn.Parameter(torch.tensor(2.0))
self.register_buffer("c", torch.tensor(1.0))
@property
def V(self) -> int:
return self.Vc
def codebook(self) -> torch.Tensor:
return self.E / self.E.norm(dim=1, keepdim=True).clamp_min(1e-8)
def masks(self) -> torch.Tensor:
return self.W / self.W.norm(dim=1, keepdim=True).clamp_min(1e-8) * math.sqrt(self.d)
def tx_frame(self, idx: torch.Tensor) -> torch.Tensor:
En = self.codebook()
e = En[idx] # (N,U,d)
m = self.masks()
return (e * m[None]).sum(dim=1) # (N,d)
def calibrate_power(self, n: int = 65536) -> None:
with torch.no_grad():
idx = torch.randint(self.Vc, (n, self.users), device=self.E.device)
y = self.tx_frame(idx)
self.c.fill_(float(y.pow(2).sum(dim=1).mean().sqrt()))
def forward(self, idx: torch.Tensor, snr_db: torch.Tensor,
h: torch.Tensor | None = None):
N = idx.shape[0]
y = self.tx_frame(idx) / self.c # (N,d)
if h is None:
h = rayleigh_gain((N, self.users), device=y.device)
sigma = snr_to_sigma2(snr_db).to(y.device).sqrt()
n = torch.randn(N, self.users, self.d, device=y.device)
y_rx = h[:, :, None] * y[:, None] + sigma[:, None, None] * n
r = y_rx / h[:, :, None].clamp_min(1e-6) # (N,U,d)
En = self.codebook()
m = self.masks()
cand = En[None, :, :] * m[:, None, :] # (U,V,d)
scores = torch.einsum("nud,uvd->nuv", r, cand)
return scores
def n_params(self) -> int:
return self.E.numel() + self.W.numel()
# ----------------------------------------------------------------------
# training
# ----------------------------------------------------------------------
def train_sse(model: SSE, iters: int = TRAIN_ITERS, batch: int = TRAIN_BATCH,
lr: float = LR, log_every: int = 0, eval_snr: float = 10.0,
eval_frames: int = 20000, seed: int = SEED):
"""Digit-wise cross entropy under superposition; cost independent of V."""
set_seed(seed)
model.to(DEVICE)
opt = torch.optim.Adam(model.parameters(), lr=lr)
ce = torch.nn.CrossEntropyLoss()
curve = []
t0 = time.time()
for it in range(1, iters + 1):
digits = torch.randint(model.vu, (batch, model.users, model.P), device=DEVICE)
snr = torch.empty(batch).uniform_(*TRAIN_SNR_DB)
model.calibrate_power(8192)
scores = model(digits, snr) * model.logit_scale.exp()
loss = ce(scores.reshape(-1, model.vu), digits.reshape(-1))
opt.zero_grad(); loss.backward(); opt.step()
if log_every and (it % log_every == 0 or it == 1):
ser = eval_ser_sse(model, [eval_snr], frames=eval_frames)[0]
curve.append((it, time.time() - t0, float(loss.detach()), ser))
model.calibrate_power()
return curve
def train_full(model: FullCodebook, iters: int = TRAIN_ITERS,
batch: int = TRAIN_BATCH, lr: float = LR, log_every: int = 0,
eval_snr: float = 10.0, eval_frames: int = 20000,
seed: int = SEED):
set_seed(seed)
model.to(DEVICE)
opt = torch.optim.Adam(model.parameters(), lr=lr)
ce = torch.nn.CrossEntropyLoss()
curve = []
t0 = time.time()
for it in range(1, iters + 1):
idx = torch.randint(model.Vc, (batch, model.users), device=DEVICE)
snr = torch.empty(batch).uniform_(*TRAIN_SNR_DB)
model.calibrate_power(8192)
scores = model(idx, snr) * model.logit_scale.exp()
loss = ce(scores.reshape(-1, model.Vc), idx.reshape(-1))
opt.zero_grad(); loss.backward(); opt.step()
if log_every and (it % log_every == 0 or it == 1):
ser = eval_ser_full(model, [eval_snr], frames=eval_frames)[0]
curve.append((it, time.time() - t0, float(loss.detach()), ser))
model.calibrate_power()
return curve
# ----------------------------------------------------------------------
# evaluation (Monte Carlo)
# ----------------------------------------------------------------------
@torch.no_grad()
def eval_ser_sse(model: SSE, snr_list, frames: int = 2_000_000,
chunk: int = 100_000, seed: int = 777):
"""Frame (vocabulary-symbol) error rate: any wrong digit is an error."""
model.eval().to(DEVICE)
out = []
for snr_db in snr_list:
g = torch.Generator(device="cpu").manual_seed(seed + int(10 * snr_db))
err = tot = 0
for n0 in range(0, frames, chunk):
n = min(chunk, frames - n0)
digits = torch.randint(model.vu, (n, model.users, model.P),
generator=g).to(DEVICE)
snr = torch.full((n,), float(snr_db))
scores = model(digits, snr)
wrong = (scores.argmax(-1) != digits).any(dim=2) # (n,U)
err += int(wrong.sum()); tot += n * model.users
out.append(err / tot)
return out
@torch.no_grad()
def eval_ser_full(model: FullCodebook, snr_list, frames: int = 2_000_000,
chunk: int = 50_000, seed: int = 777):
model.eval().to(DEVICE)
if model.Vc >= 4096:
chunk = max(512, (1 << 23) // model.Vc)
out = []
for snr_db in snr_list:
g = torch.Generator(device="cpu").manual_seed(seed + int(10 * snr_db))
err = tot = 0
for n0 in range(0, frames, chunk):
n = min(chunk, frames - n0)
idx = torch.randint(model.Vc, (n, model.users), generator=g).to(DEVICE)
snr = torch.full((n,), float(snr_db))
scores = model(idx, snr)
err += int((scores.argmax(-1) != idx).sum()); tot += n * model.users
out.append(err / tot)
return out
# ----------------------------------------------------------------------
# conventional digital scheme: OMA with BPSK per real dimension (QPSK
# per complex subcarrier under the declared real-imaginary stacking)
# ----------------------------------------------------------------------
def oma_ser(snr_db_list, bits: int = 16, n_grid: int = 200_000):
"""Semi-analytic expression under the real-dimension convention.
All bits of a vocabulary symbol share the same flat-fading gain, so
SER = E_h[1 - (1 - Q(h sqrt(snr)))^bits] with h^2 ~ Exp(1); the
expectation is evaluated by numerical integration on a dense grid."""
x = (np.arange(n_grid) + 0.5) / n_grid # uniform quantiles
h = np.sqrt(-np.log(1.0 - x)) # inverse-CDF transform
out = []
for s in snr_db_list:
g = 10.0 ** (s / 10.0)
q = 0.5 * np.array([math.erfc(v / math.sqrt(2.0)) for v in
np.clip(h * math.sqrt(g), 0, 38)])
out.append(float(np.mean(1.0 - (1.0 - q) ** bits)))
return out
@torch.no_grad()
def oma_ser_mc(snr_db_list, bits: int = 16, frames: int = 2_000_000,
chunk: int = 200_000, seed: int = 777):
"""Monte Carlo check of the closed form (same channel conventions)."""
out = []
for s in snr_db_list:
sigma = math.sqrt(1.0 / (10.0 ** (s / 10.0))) # per-dim, unit Es
err = tot = 0
torch.manual_seed(seed + int(10 * s))
for n0 in range(0, frames, chunk):
n = min(chunk, frames - n0)
h = rayleigh_gain((n, 1))
b = torch.randint(0, 2, (n, bits), device=DEVICE) * 2.0 - 1.0
y = h * b + sigma * torch.randn(n, bits, device=DEVICE)
wrong = ((y * b) < 0).any(dim=1)
err += int(wrong.sum()); tot += n
out.append(err / tot)
return out
# ----------------------------------------------------------------------
# analysis: union bound with residual-interference Gaussian approximation
# ----------------------------------------------------------------------
@torch.no_grad()
def sse_union_bound(model: SSE, snr_db_list, n_mc_h: int = 200_000):
"""P_e <= 1 - (1 - min(1, sum_pairs)) ... digit union bound averaged
over the Rayleigh gain, with measured residual interference treated
as additional Gaussian noise (approximation stated in the paper)."""
model.eval().to(DEVICE)
Bn = model.unit_codebook()
m = model.masks()
c = float(model.c)
# residual interference power per dimension at user u (measured)
digits = torch.randint(model.vu, (65536, model.users, model.P), device=DEVICE)
e = Bn[digits] / math.sqrt(model.P)
x = e * m[None, :, None, :] # (N,U,P,L)
y = x.sum(dim=1) / c # (N,P,L)
# per-user: signal = x_u/c, interference = (y - x_u/c)
interf_pw = []
dmin2 = []
for u in range(model.users):
su = x[:, u] / c # (N,P,L)
iu = y - su
# project interference onto the normalized candidate directions
cand = Bn * m[u][None, :] # (Vu,L)
cn = cand / cand.norm(dim=1, keepdim=True).clamp_min(1e-8)
proj = torch.einsum("npl,vl->npv", iu, cn)
interf_pw.append(float(proj.pow(2).mean()))
# pairwise distances of the scaled candidate set (tx side scaling)
cs = cand / (c * math.sqrt(model.P))
dd = torch.cdist(cs, cs)
dmin2.append(float((dd + torch.eye(model.vu, device=DEVICE) * 1e9).min() ** 2))
res = []
g = rayleigh_gain(n_mc_h)
for s in snr_db_list:
sig2 = float(snr_to_sigma2(torch.tensor(s)))
pe_users = []
for u in range(model.users):
# effective noise per dim after equalization: sig2/h^2 + interf
sig_eff2 = sig2 / g.pow(2) + interf_pw[u]
arg = torch.sqrt(torch.clamp(torch.tensor(dmin2[u], device=DEVICE)
/ (4.0 * sig_eff2), min=0.0))
q = 0.5 * torch.erfc(arg / math.sqrt(2.0))
p_digit = torch.clamp((model.vu - 1) * q, max=1.0)
p_frame = 1.0 - (1.0 - p_digit) ** model.P
pe_users.append(float(p_frame.mean()))
res.append(sum(pe_users) / len(pe_users))
return res
def write_csv(path: Path, header: list[str], rows) -> None:
with open(path, "w") as f:
f.write(",".join(header) + "\n")
for row in rows:
f.write(",".join(f"{v:.10g}" if isinstance(v, float) else str(v)
for v in row) + "\n")
print("[csv]", path)
code/verify_math.py
+201
View File
@@ -0,0 +1,201 @@
"""Numerical verification of the closed forms in Sections IV and V of
paper 11 (mask-as-key encryption and jamming robustness).
Every claim that enters the manuscript is checked here against Monte
Carlo, with a PASS/FAIL verdict and the achieved agreement level printed.
Real-vector convention, dimension d, U users, codebook of V unit-norm
codewords, per-user masks with zero-mean entries normalized to
||m||^2 = d (so E[m_k^2] = 1).
Notation matches the tex:
y = (1/c) sum_u e_{s_u} .* m_u transmit frame
legit score z_{u,i} = r_u^T (e_i .* m_u), r_u = y + n_u/h_u
eve score zE_{u,i} = (y_E/h_E)^T (e_i .* mtil_u)
jammer adds h_J sqrt(rho) w to the victim observation
Run on CPU (NumPy); no training involved, pure algebra checks.
"""
from __future__ import annotations
import numpy as np
RNG = np.random.default_rng(2026)
D, U, V = 64, 4, 256
def unit_codebook(V, d, rng):
E = rng.standard_normal((V, d))
return E / np.linalg.norm(E, axis=1, keepdims=True)
def masks(U, d, rng):
"""Zero-mean entries normalized so ||m_u||^2 = d."""
M = rng.standard_normal((U, d))
return M / np.linalg.norm(M, axis=1, keepdims=True) * np.sqrt(d)
def report(tag, claim, emp, tol, extra=""):
err = abs(claim - emp)
ok = err <= tol
print(f"[{'PASS' if ok else 'FAIL'}] {tag}: claim={claim:.5g} "
f"emp={emp:.5g} |err|={err:.2g} tol={tol:g} {extra}")
return ok
def v1_legit_self_alignment():
"""Claim: E[e_s^T diag(m^2) e_s] = 1 (signal self-correlation)."""
E = unit_codebook(V, D, RNG)
vals = []
for _ in range(4000):
m = masks(1, D, RNG)[0]
s = RNG.integers(V)
vals.append(float((E[s] ** 2) @ (m ** 2)))
return report("V1 legit self-alignment", 1.0, float(np.mean(vals)), 2e-2)
def v2_eve_uninformed():
"""Claim: with an independent substitute mask, the eavesdropper's
correct-index correlation has the same mean as any wrong index, so
the mean advantage is zero and the eavesdropper SER = (V-1)/V,
independent of SNR."""
E = unit_codebook(V, D, RNG)
# mean advantage of the true index over the wrong indices, noiseless
adv = []
ser_by_snr = {}
for snr_db in [0.0, 10.0, 20.0, 80.0]: # 80 dB stands in for noiseless
sigma = np.sqrt(1.0 / (D * 10 ** (snr_db / 10.0)))
err = 0
trials = 6000
for _ in range(trials):
s = RNG.integers(V, size=U)
M = masks(U, D, RNG)
c = 1.0 # scale-invariant for argmax
y = np.zeros(D)
for u in range(U):
y += E[s[u]] * M[u]
y /= np.sqrt(U) # any fixed scale
# eavesdropper targets user 0 with an independent wrong mask
mtil = masks(1, D, RNG)[0]
hE = np.sqrt(-np.log(RNG.random()))
rE = y + (sigma / hE) * RNG.standard_normal(D)
scores = (E * mtil) @ rE # (V,)
if snr_db == 80.0:
adv.append(scores[s[0]] - scores.mean())
if scores.argmax() != s[0]:
err += 1
ser_by_snr[snr_db] = err / trials
chance = (V - 1) / V
ok1 = report("V2a eve mean advantage", 0.0, float(np.mean(adv)), 3e-3)
ok2 = True
for snr_db, ser in ser_by_snr.items():
tag = f"V2b eve SER @ {int(snr_db)}dB"
ok2 &= report(tag, chance, ser, 1.5e-2)
return ok1 and ok2
def v3_leakage_vs_correlation():
"""Claim: if the substitute mask has normalized correlation
rho = <m,mtil>/d with the true mask, the eavesdropper's true-index
bias grows linearly in rho; independent random masks give
E|rho| = O(1/sqrt(d)); orthogonal masks give rho = 0."""
E = unit_codebook(V, D, RNG)
# (a) bias vs prescribed rho
slopes = []
for rho in [0.0, 0.25, 0.5, 0.75, 1.0]:
bias = []
for _ in range(3000):
m = masks(1, D, RNG)[0]
mp = masks(1, D, RNG)[0]
mp = mp - (mp @ m) / (m @ m) * m # orthogonalize
mp = mp / np.linalg.norm(mp) * np.sqrt(D)
mtil = rho * m + np.sqrt(1 - rho ** 2) * mp
s = RNG.integers(V)
# noiseless single-user useful alignment for the true index
bias.append(float((E[s] ** 2) @ (m * mtil)))
slopes.append((rho, float(np.mean(bias))))
# claim: bias(rho) = rho * bias(1); check linearity
b1 = slopes[-1][1]
lin_ok = all(abs(b - rho * b1) <= 3e-2 for rho, b in slopes)
print(f"[{'PASS' if lin_ok else 'FAIL'}] V3a bias linear in rho: "
+ ", ".join(f"rho={r:.2f}->{b:.3f}" for r, b in slopes))
# (b) random independent mask correlation: E|corr| = sqrt(2/(pi d))
# (the folded-normal mean of a N(0, 1/d) variable)
corrs = []
for _ in range(5000):
m = masks(1, D, RNG)[0]
mt = masks(1, D, RNG)[0]
corrs.append(abs((m @ mt) / D))
emp = float(np.mean(corrs))
claim = float(np.sqrt(2.0 / (np.pi * D)))
ok_b = report("V3b random mask E|corr|", claim, emp, 0.1 * claim)
return lin_ok and ok_b
def v4_blind_jammer_spread():
"""Claim: a mask-blind jammer (w independent of m_u) contributes a
zero-mean term to every candidate score with variance
(hJ^2/hu^2) rho * sum_k w_k^2 e_{i,k}^2, i.e. it is spread with no
systematic bias toward any index."""
E = unit_codebook(V, D, RNG)
rho = 1.0
# (a) structural claim: the mask projection g_i = (w .* m)^T e_i is
# zero-mean, decoupled from the positive gain factor hJ/hu.
g, var_emp, var_cl = [], [], []
for _ in range(20000):
m = masks(1, D, RNG)[0]
w = RNG.standard_normal(D); w /= np.linalg.norm(w)
i = RNG.integers(V)
gi = (w * m) @ E[i]
g.append(gi)
var_emp.append(gi ** 2)
var_cl.append(np.sum(w ** 2 * E[i] ** 2))
ok1 = report("V4a blind jammer projection mean", 0.0,
float(np.mean(g)), 3e-3)
# (b) variance of the projection matches sum_k w_k^2 e_{i,k}^2; the
# full contribution scales this by (hJ^2/hu^2) rho.
ok2 = report("V4b blind jammer projection variance",
float(np.mean(var_cl)), float(np.mean(var_emp)),
0.05 * float(np.mean(var_cl)))
return ok1 and ok2
def v5_matched_jammer_concentrates():
"""Claim: a mask-matched jammer aligned with the victim key for a
target index t creates a bias of order one on index t, while the
blind-jammer projection has zero mean and RMS of order 1/sqrt(d).
The physically meaningful separation is matched bias over blind RMS,
which is sqrt(d) (the sample mean of the blind bias estimates zero
and is pure Monte Carlo noise, so it is NOT a valid denominator)."""
E = unit_codebook(V, D, RNG)
bias_matched, blind_sq = [], []
for _ in range(3000):
m = masks(1, D, RNG)[0]
t = RNG.integers(V)
wm = E[t] * m; wm /= np.linalg.norm(wm) # matched (needs m)
wb = RNG.standard_normal(D); wb /= np.linalg.norm(wb) # blind
bias_matched.append(float((wm * m) @ E[t]))
blind_sq.append(float(((wb * m) @ E[t]) ** 2))
bm = float(np.mean(bias_matched))
brms = float(np.sqrt(np.mean(blind_sq)))
ratio = bm / brms
ok = abs(ratio - np.sqrt(D)) <= 0.25 * np.sqrt(D) and bm > 0.9
print(f"[{'PASS' if ok else 'FAIL'}] V5 matched bias / blind RMS: "
f"matched={bm:.3f} blind_rms={brms:.4f} ratio={ratio:.1f} "
f"(claim sqrt(d)={np.sqrt(D):.1f})")
return ok
def main():
print(f"config d={D} U={U} V={V}\n")
results = {
"V1": v1_legit_self_alignment(),
"V2": v2_eve_uninformed(),
"V3": v3_leakage_vs_correlation(),
"V4": v4_blind_jammer_spread(),
"V5": v5_matched_jammer_concentrates(),
}
print("\nsummary:", {k: ("PASS" if v else "FAIL") for k, v in results.items()})
print("ALL PASS" if all(results.values()) else "SOME FAILED")
if __name__ == "__main__":
main()