Skip to content

Instantly share code, notes, and snippets.

@lastforkbender

lastforkbender/rd2_1331.py

Created June 25, 2026 02:56
Show Gist options

  • Select an option

    Learn more about clone URLs
  • Save lastforkbender/bcdaaba5784d0ff4ee36b3d9149afc55 to your computer and use it in GitHub Desktop.

Select an option

Learn more about clone URLs
Save lastforkbender/bcdaaba5784d0ff4ee36b3d9149afc55 to your computer and use it in GitHub Desktop.
Download ZIP
RD2/1331
Raw
rd2_1331.py
----------------------------------------------------------------------------------
# RD2/1331/frostman_measure.py
from __future__ import annotations
import os
import math
import logging
import itertools
from dataclasses import dataclass, field
from typing import Sequence
import numpy as np
from scipy.optimize import minimize_scalar
from sobolev_padding_pipeline import ncd, aes_encrypt, MetricSubset, box_counting_dimension
logger = logging.getLogger(__name__)
logging.basicConfig(level=logging.INFO, format="%(levelname)s | %(message)s")
DEFAULT_TOL_MU: float = 1e-6
DEFAULT_TOL_N: float = 0.5
DEFAULT_TOL_DIM: float = 0.05
DEFAULT_MAX_ITER: int = 200
DEFAULT_C_SEARCH: bool = True
def build_distance_matrix(elements: list[bytes]) -> np.ndarray:
n = len(elements); D = np.zeros((n, n), dtype=np.float64)
for i, j in itertools.combinations(range(n), 2):
d = ncd(elements[i], elements[j])
D[i, j] = d; D[j, i] = d
logger.info("Distance matrix built: %d×%d", n, n)
return D
def seed_exponent(elements: list[bytes], D: np.ndarray, mu: np.ndarray,) -> tuple[float, int]:
n = len(elements); diam = float(np.max(D))
if diam == 0.0:
logger.warning("Diameter = 0; all elements identical. s₀ = 0.")
return 0.0, 0
flat_idx = int(np.argmax(D))
i_star = flat_idx // n; j_star = flat_idx % n
scores = np.maximum(D[:, i_star], D[:, j_star])
x_star = int(np.argmin(scores))
eps_ref = diam / 2.0
in_ball = D[x_star, :] <= eps_ref; mu_ball = float(np.sum(mu[in_ball]))
if mu_ball <= 0.0 or eps_ref <= 0.0:
logger.warning("Degenerate seed: mu_ball=%.6f eps_ref=%.6f; s₀=1.0", mu_ball, eps_ref)
return 1.0, x_star
s0 = -math.log(mu_ball) / math.log(eps_ref); s0 = max(0.0, min(s0, 2.0))
logger.info("Diameter seed: diam=%.4f x*=%d ε_ref=%.4f μ(B)=%.6f s₀=%.4f", diam, x_star, eps_ref, mu_ball, s0,)
return s0, x_star
def ball_masses(D: np.ndarray, mu: np.ndarray, eps: float,) -> np.ndarray:
in_ball = D <= eps
return in_ball @ mu
def update_exponent(D: np.ndarray, mu: np.ndarray, eps_grid: np.ndarray,) -> float:
n = D.shape[0]; local_s: list[float] = []
for i in range(n):
row_s: list[float] = []
for eps in eps_grid:
m_ball = float(ball_masses(D, mu, eps)[i])
if 0.0 < m_ball < 1.0 and eps > 0.0:
s_local = -math.log(m_ball) / math.log(eps)
if math.isfinite(s_local) and s_local > 0.0: row_s.append(s_local)
if row_s: local_s.append(float(np.median(row_s)))
if not local_s:
return 0.0
s_new = float(np.median(local_s))
return max(0.0, min(s_new, 2.0))
def update_measure(D: np.ndarray, mu: np.ndarray, s: float, eps: float,) -> np.ndarray:
masses = ball_masses(D, mu, eps); target = eps ** s
weights = np.where(masses > 0.0, target / masses, 1.0)
mu_new = mu * weights; total = mu_new.sum()
if total <= 0.0:
return mu.copy()
return mu_new / total
def covering_number_from_D(D: np.ndarray, eps: float) -> int:
n = D.shape[0]; covered = np.zeros(n, dtype=bool); n_centres = 0
for i in range(n):
if not covered[i]:
n_centres += 1; covered |= (D[i, :] <= eps)
return n_centres
def fit_frostman_constant(D: np.ndarray, mu: np.ndarray, s: float, eps_grid: np.ndarray,) -> float:
C_vals: list[float] = []
for eps in eps_grid:
if eps <= 0.0:
continue
masses = ball_masses(D, mu, eps); target = eps ** s
if target > 0.0: C_vals.extend((masses / target).tolist())
return float(np.max(C_vals)) if C_vals else 1.0
@dataclass
class VerificationResult:
frostman_holds: bool
dimension_coincides: bool
C: float
s_final: float
dim_B: float
dim_gap: float
max_violation: float
tol_dim: float
n_balls_checked: int
def certificate(self) -> str:
lines = ["", "=" * 62, " FROSTMAN MEASURE CERTIFICATE", "=" * 62,
f" Frostman exponent s = {self.s_final:.6f}",
f" Frostman constant C = {self.C:.6f}",
f" Box-counting dim_B = {self.dim_B:.6f}",
f" Dimension gap |s-dim_B|= {self.dim_gap:.6f} (tol={self.tol_dim})",
f" Max ball violation = {self.max_violation:.2e}",
f" Balls checked = {self.n_balls_checked}", "",
" FROSTMAN CONDITION μ(B(x,ε)) ≤ C·ε^s :",
f" {'✓ HOLDS' if self.frostman_holds else '✗ VIOLATED'}", "",
" DIMENSION COINCIDENCE dim_H(F) = dim_B(F) :",
f" {'✓ CONFIRMED' if self.dimension_coincides else '✗ NOT CONFIRMED'}", "",]
if self.frostman_holds and self.dimension_coincides:
lines += [" CONCLUSION:",
" NCD balls form a valid covering family for F.",
" The Hausdorff and box-counting dimensions coincide.",
" Pipeline covering argument is formally closed.",]
elif self.frostman_holds:
lines += [" CONCLUSION:",
" Frostman condition holds but dimension gap exceeds tolerance.",
" Increase iterations or tighten eps_grid for dimension claim.",]
else:
lines += [" CONCLUSION:",
" Frostman condition violated — increase max_iter or",
" check that F has sufficient cardinality (|F| ≥ 8 recommended).",]
lines.append("=" * 62)
return "\n".join(lines)
def verify_terminal(D: np.ndarray, mu: np.ndarray, s: float, dim_B: float, eps_grid: np.ndarray, tol_dim: float = DEFAULT_TOL_DIM,) -> VerificationResult:
C = fit_frostman_constant(D, mu, s, eps_grid); n = D.shape[0]; max_viol = 0.0; n_checked = 0
for eps in eps_grid:
if eps <= 0.0:
continue
masses = ball_masses(D, mu, eps); allowed = C * (eps ** s); viols = masses - allowed
max_viol = max(max_viol, float(np.max(viols))); n_checked += n
frostman_holds = max_viol <= 1e-9; dim_gap = abs(s - dim_B); dimension_coincides = dim_gap < tol_dim
return VerificationResult(frostman_holds=frostman_holds, dimension_coincides=dimension_coincides, C=C, s_final=s, dim_B=dim_B, dim_gap=dim_gap, max_violation=max_viol, tol_dim=tol_dim, n_balls_checked=n_checked,)
@dataclass
class FrostmanResult:
mu: np.ndarray
s: float
s_history: list[float]
mu_delta_history: list[float]
N_history: list[int]
invariant_ratio_history: list[float]
n_iter: int
converged: bool
verification: VerificationResult
def invariant_ratio(self) -> float | None:
if not self.invariant_ratio_history:
return None
return self.invariant_ratio_history[-1]
def construct_frostman(subset: MetricSubset, dim_B: float, tol_mu: float = DEFAULT_TOL_MU, tol_N: float = DEFAULT_TOL_N, tol_dim: float = DEFAULT_TOL_DIM, max_iter: int = DEFAULT_MAX_ITER, n_eps: int = 12,) -> FrostmanResult:
elements = subset.elements; m = len(elements)
if m == 0:
_v = VerificationResult(frostman_holds=True, dimension_coincides=True, C=0.0, s_final=float("-inf"), dim_B=float("-inf"), dim_gap=0.0, max_violation=0.0, tol_dim=tol_dim, n_balls_checked=0,)
return FrostmanResult(mu=np.array([]), s=float("-inf"), s_history=[], mu_delta_history=[], invariant_ratio_history=[], N_history=[], n_iter=0, converged=True, verification=_v,)
if m == 1:
_v = VerificationResult(frostman_holds=True, dimension_coincides=True, C=1.0, s_final=0.0, dim_B=0.0, dim_gap=0.0, max_violation=0.0, tol_dim=tol_dim, n_balls_checked=1,)
return FrostmanResult(mu=np.array([1.0]), s=0.0, s_history=[0.0], mu_delta_history=[], N_history=[1], n_iter=0, converged=True, verification=_v,)
D = build_distance_matrix(elements); diam = float(np.max(D))
if diam == 0.0:
_mu = np.ones(m, dtype=np.float64) / m
_v = VerificationResult(frostman_holds=True, dimension_coincides=True, C=1.0, s_final=0.0, dim_B=0.0, dim_gap=0.0, max_violation=0.0, tol_dim=tol_dim, n_balls_checked=m,)
return FrostmanResult(mu=_mu, s=0.0, s_history=[0.0], mu_delta_history=[], N_history=[1], n_iter=0, converged=True, verification=_v,)
if diam > 1.1:
raise ValueError(f"F appears unbounded: diam(F)={diam:.4f} > 1.1. "
"NCD should be ≤ 1+δ for any valid compressor. "
"Inspect _compressed_len() for anomalous output.")
eps_lo = diam / (2 ** n_eps); eps_hi = diam / 2.0; eps_grid = np.geomspace(eps_lo, eps_hi, num=n_eps)
mu = np.ones(m, dtype=np.float64) / m
s, x_star = seed_exponent(elements, D, mu)
logger.info("Initial exponent s₀ = %.6f", s)
eps_work = float(np.sqrt(eps_lo * eps_hi))
s_history: list[float] = [s]
mu_delta_history: list[float] = []
invariant_ratio_history: list[float] = []
N_history: list[int] = [covering_number_from_D(D, eps_work)]
converged = False
for k in range(max_iter):
mu_new = update_measure(D, mu, s, eps_work); mu_delta = float(np.linalg.norm(mu_new - mu))
mu_norm = float(np.linalg.norm(mu))
inv_ratio = mu_delta / mu_norm if mu_norm > 0.0 else 0.0
s_new = update_exponent(D, mu_new, eps_grid)
N_new = covering_number_from_D(D, eps_work)
N_old = N_history[-1]
mu_delta_history.append(mu_delta)
invariant_ratio_history.append(inv_ratio)
N_history.append(N_new); s_history.append(s_new)
logger.debug("iter %3d | s=%.6f→%.6f | Δμ=%.2e | ratio=%.4f | N=%d→%d", k + 1, s, s_new, mu_delta, inv_ratio, N_old, N_new,)
mu = mu_new; s = s_new
mu_ok = mu_delta < tol_mu
N_ok = abs(N_new - N_old) < tol_N
if mu_ok and N_ok:
logger.info("Converged at iteration %d: Δμ=%.2e < %.2e, ΔN=%d < %.1f", k + 1, mu_delta, tol_mu, abs(N_new - N_old), tol_N,)
converged = True; n_iter = k + 1
break
else:
logger.warning("Did not converge within %d iterations.", max_iter)
n_iter = max_iter
logger.info("Running terminal verification…")
verification = verify_terminal(D, mu, s, dim_B, eps_grid, tol_dim)
logger.info("Frostman: %s | Dimension coincidence: %s", "✓" if verification.frostman_holds else "✗", "✓" if verification.dimension_coincides else "✗",)
return FrostmanResult(mu=mu, s=s, s_history=s_history, mu_delta_history=mu_delta_history, invariant_ratio_history=invariant_ratio_history, N_history=N_history, n_iter=n_iter, converged=converged, verification=verification,)
def frostman_lambda(result: FrostmanResult) -> float:
lam = float(np.clip(result.s, 0.0, 1.0))
return lam
if __name__ == "__main__":
print("Generating demo ciphertexts for Frostman construction…")
key = os.urandom(16)
plaintexts = [b"Alpha block data for Frostman.",
b"Beta block data for Frostman.",
b"Gamma block for coverage test.",
b"Delta padding error injection.",
b"Epsilon another test string xx",
b"Zeta final element in set F.",]
ciphertexts = []
for pt in plaintexts:
ct, iv = aes_encrypt(pt, key); ciphertexts.append(ct)
buf = bytearray(ciphertexts[3]); buf[-1] ^= 0xFF; ciphertexts[3] = bytes(buf)
subset = MetricSubset(elements=ciphertexts)
print("Computing box-counting dimension…")
dim_B, _, _ = box_counting_dimension(subset)
print(f" dim_B(F) = {dim_B:.4f}")
print("Constructing Frostman measure…")
result = construct_frostman(subset, dim_B)
print(result.verification.certificate())
lam = frostman_lambda(result)
print(f"\nPipeline λ_effective = {lam:.4f}")
print(f"Converged: {result.converged} in {result.n_iter} iterations")
print(f"Invariant ratio (terminal): {result.invariant_ratio()}")
----------------------------------------------------------------------------------
# RD2/1331/sobolev_padding_pipeline.py
from __future__ import annotations
import os
import zlib
import math
import logging
import itertools
from dataclasses import dataclass, field
from typing import Sequence
import numpy as np
from scipy.optimize import least_squares
from Crypto.Cipher import AES
from Crypto.Util.Padding import pad, unpad
logger = logging.getLogger(__name__)
logging.basicConfig(level=logging.INFO, format="%(levelname)s | %(message)s")
def _compressed_len(data: bytes) -> int:
return len(zlib.compress(data, level=9))
def ncd(x: bytes, y: bytes) -> float:
if x == y:
return 0.0
cx = _compressed_len(x); cy = _compressed_len(y); cxy = _compressed_len(x + y)
return (cxy - min(cx, cy)) / max(cx, cy)
@dataclass
class MetricSubset:
elements: list[bytes]
def __post_init__(self) -> None:
pass
def diameter(self) -> float:
diam = 0.0
for x, y in itertools.combinations(self.elements, 2): diam = max(diam, ncd(x, y))
return diam
def covering_number(self, eps: float) -> int:
uncovered = list(self.elements); centres: list[bytes] = []
while uncovered:
c = uncovered.pop(0); centres.append(c)
uncovered = [u for u in uncovered if ncd(c, u) > eps]
return len(centres)
def box_counting_dimension(
subset: MetricSubset,
eps_values: Sequence[float] | None = None,) -> tuple[float, np.ndarray, np.ndarray]:
diam = subset.diameter()
if diam == 0.0:
logger.warning("All elements identical; dim_B = 0.")
return 0.0, np.array([]), np.array([])
if eps_values is None: eps_values = [diam * (0.5 ** k) for k in range(1, 7)]
log_eps_arr: list[float] = []; log_N_arr: list[float] = []
for eps in eps_values:
N = subset.covering_number(eps)
if N > 0 and eps > 0: log_eps_arr.append(-math.log(eps)); log_N_arr.append(math.log(N))
if len(log_eps_arr) < 2:
logger.warning("Insufficient ε samples for regression.")
return 0.0, np.array(log_eps_arr), np.array(log_N_arr)
x = np.array(log_eps_arr); y = np.array(log_N_arr)
A = np.column_stack([x, np.ones_like(x)])
coeffs, _, _, _ = np.linalg.lstsq(A, y, rcond=None)
dim_B = float(coeffs[0])
logger.info("Box-counting dimension dim_B(F) ≈ %.4f", dim_B)
return dim_B, x, y
BLOCK_SIZE = AES.block_size
def aes_encrypt(plaintext: bytes, key: bytes, iv: bytes | None = None) -> tuple[bytes, bytes]:
if iv is None: iv = os.urandom(BLOCK_SIZE)
padded = pad(plaintext, BLOCK_SIZE, style="pkcs7")
cipher = AES.new(key, AES.MODE_CBC, iv)
return cipher.encrypt(padded), iv
def aes_decrypt_raw(ciphertext: bytes, key: bytes, iv: bytes) -> bytes:
cipher = AES.new(key, AES.MODE_CBC, iv)
return cipher.decrypt(ciphertext)
@dataclass
class PaddingError:
block_index: int
observed_byte: int
expected_pad: int
error_vector: np.ndarray
severity: float
def pkcs7_validate(raw_padded: bytes) -> list[PaddingError]:
errors: list[PaddingError] = []; n_blocks = len(raw_padded) // BLOCK_SIZE
for b in range(n_blocks):
block = raw_padded[b * BLOCK_SIZE : (b + 1) * BLOCK_SIZE]
last_byte = block[-1]
pad_val = last_byte if 1 <= last_byte <= BLOCK_SIZE else 0
expected = np.zeros(BLOCK_SIZE, dtype=np.int32)
if pad_val > 0: expected[-pad_val:] = pad_val
observed = np.array(list(block), dtype=np.int32)
residual = observed - expected
tail = block[-pad_val:] if pad_val > 0 else block[-1:]
if pad_val == 0 or not all(byte == pad_val for byte in tail):
severity = float(np.linalg.norm(residual))
errors.append(PaddingError(block_index=b, observed_byte=last_byte, expected_pad=pad_val, error_vector=residual, severity=severity,))
logger.debug("Padding error in block %d: last_byte=%d expected_pad=%d severity=%.3f", b, last_byte, pad_val, severity,)
return errors
@dataclass
class CorrectionResult:
corrected_raw: bytes
corrected_plain: bytes | None
residuals: list[np.ndarray]
ls_cost: float
dim_B: float
errors_found: list[PaddingError]
errors_corrected: int
def _ls_block_correction(block: bytes, pad_error: PaddingError, dim_B: float,) -> bytes:
observed = np.array(list(block), dtype=np.float64)
b_vec = pad_error.error_vector.astype(np.float64)
lambda_reg = np.clip(dim_B, 0.0, 1.0)
def residual_fn(delta: np.ndarray) -> np.ndarray:
data_term = delta - b_vec; reg_term = math.sqrt(lambda_reg) * delta
return np.concatenate([data_term, reg_term])
x0 = np.zeros(BLOCK_SIZE, dtype=np.float64)
result = least_squares(residual_fn, x0, method="lm")
corrected_floats = observed - result.x
corrected_ints = np.clip(np.round(corrected_floats), 0, 255).astype(np.uint8)
return bytes(corrected_ints)
def least_squares_correction(raw_padded: bytes, errors: list[PaddingError], dim_B: float,) -> tuple[bytes, list[np.ndarray], float]:
buf = bytearray(raw_padded); residuals: list[np.ndarray] = []; total_cost = 0.0
for err in errors:
start = err.block_index * BLOCK_SIZE
block = bytes(buf[start : start + BLOCK_SIZE])
fixed = _ls_block_correction(block, err, dim_B)
buf[start : start + BLOCK_SIZE] = fixed
post_residual = np.array(list(fixed), dtype=np.float64) - np.array(list(block), dtype=np.float64)
residuals.append(post_residual)
total_cost += float(np.sum(post_residual ** 2))
return bytes(buf), residuals, math.sqrt(total_cost)
def run_pipeline(ciphertexts: list[bytes], key: bytes, ivs: list[bytes], eps_values: Sequence[float] | None = None,) -> list[CorrectionResult]:
if len(ciphertexts) != len(ivs):
raise ValueError("ciphertexts and ivs must have the same length.")
logger.info("Building metric subset F with %d elements.", len(ciphertexts))
subset = MetricSubset(elements=list(ciphertexts))
logger.info("Computing box-counting dimension over (X, d_NCD)…")
dim_B, log_eps, log_N = box_counting_dimension(subset, eps_values)
results: list[CorrectionResult] = []
for idx, (ct, iv) in enumerate(zip(ciphertexts, ivs)):
logger.info("--- Processing ciphertext %d / %d ---", idx + 1, len(ciphertexts))
try:
raw_padded = aes_decrypt_raw(ct, key, iv)
except Exception as exc:
logger.error("Decryption failed for ciphertext %d: %s", idx, exc)
continue
errors = pkcs7_validate(raw_padded); logger.info("Detected %d padding error(s).", len(errors))
if not errors:
try:
plain = unpad(raw_padded, BLOCK_SIZE, style="pkcs7")
except ValueError:
plain = None
results.append(CorrectionResult(corrected_raw=raw_padded, corrected_plain=plain, residuals=[], ls_cost=0.0, dim_B=dim_B, errors_found=[], errors_corrected=0,))
continue
corrected_raw, residuals, ls_cost = least_squares_correction(raw_padded, errors, dim_B)
logger.info("LS correction complete. Cost: %.6f", ls_cost)
try:
corrected_plain = unpad(corrected_raw, BLOCK_SIZE, style="pkcs7")
logger.info("Unpadding successful after correction.")
except ValueError as exc:
logger.warning("Unpadding still failed after correction: %s", exc)
corrected_plain = None
results.append(CorrectionResult(corrected_raw=corrected_raw, corrected_plain=corrected_plain, residuals=residuals, ls_cost=ls_cost, dim_B=dim_B, errors_found=errors, errors_corrected=len(errors),))
return results
def report(results: list[CorrectionResult]) -> None:
print("\n" + "=" * 60); print(" SOBOLEV PADDING PIPELINE — RESULTS SUMMARY"); print("=" * 60)
for i, r in enumerate(results):
print(f"\n[Ciphertext {i + 1}]")
print(f" Box-counting dim_B(F) : {r.dim_B:.4f}")
print(f" Padding errors found : {len(r.errors_found)}")
print(f" Errors corrected : {r.errors_corrected}")
print(f" LS correction cost : {r.ls_cost:.6f}")
if r.corrected_plain is not None: print(f" Recovered plaintext : {r.corrected_plain!r}")
else: print(f" Recovered plaintext : [uncorrectable — manual inspection needed]")
for err in r.errors_found:
print(f" ↳ Block {err.block_index:2d} "
f"last_byte={err.observed_byte} "
f"expected_pad={err.expected_pad} "
f"severity={err.severity:.3f}")
print("\n" + "=" * 60)
if __name__ == "__main__":
print("Generating demo ciphertexts…")
key = os.urandom(16)
plaintexts = [b"Hello, Cool Beans! This is a roger clean beans 10-4.",
b"Boundary block test: 16 bytes! ",
b"Short8-2.",
b"Message received, clean roger 482916336381063051854.",]
ciphertexts = []; ivs = []
for pt in plaintexts:
ct, iv = aes_encrypt(pt, key); ciphertexts.append(ct); ivs.append(iv)
corrupted = bytearray(ciphertexts[1])
corrupted[-1] ^= 0xFF
ciphertexts[1] = bytes(corrupted)
print("Injected padding error into ciphertext 2.")
results = run_pipeline(ciphertexts, key, ivs)
report(results)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment