Aethel Noise

The gate is driven by jitter entropy rather than voltage. A sample above the adaptive threshold conducts; below it, the channel is closed.

import time, socket, statistics

class NoiseSource:
    """Samples raw timing noise from a network path."""
    def __init__(self, host="1.1.1.1", port=80, window=64):
        self.host, self.port, self.window = host, port, window
        self.history = []

    def sample(self):
        t0 = time.perf_counter_ns()
        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        s.settimeout(1.0)
        try:
            s.connect_ex((self.host, self.port))
        finally:
            s.close()
        rtt = time.perf_counter_ns() - t0
        self.history.append(rtt)
        if len(self.history) > self.window:
            self.history.pop(0)
        return rtt

    def jitter(self):
        if len(self.history) < 4:
            return 0.0
        d = [abs(b - a) for a, b in zip(self.history, self.history[1:])]
        return statistics.fmean(d)


class NoiseTransistor:
    """Gate conducts when noise entropy crosses an adaptive threshold."""
    def __init__(self, source, bias=1.0):
        self.src = source
        self.bias = bias
        self.threshold = None

    def _calibrate(self):
        j = [self.src.jitter() for _ in range(16) if self.src.sample()]
        j = [x for x in j if x > 0]
        self.threshold = statistics.median(j) * self.bias if j else 1.0

    def conduct(self, gate=True):
        if self.threshold is None:
            self._calibrate()
        self.src.sample()
        if not gate:
            return 0
        return 1 if self.src.jitter() > self.threshold else 0


if __name__ == "__main__":
    t = NoiseTransistor(NoiseSource())
    print("noise-gated output:", [t.conduct() for _ in range(20)])

		 

Gates are built from noise transistors, then composed into a ripple-carry adder. Correctness is fully deterministic, yet the execution time is bound to physical entropy: every gate blocks and polls the noise transistor dynamically until it stabilizes and confirms the required quantum state. It works. It is roughly ten million times slower than a pocket calculator, which is precisely the result that redirected the whole program towards listening.

from noise_transistor import NoiseTransistor, NoiseSource

class NoiseLogic:
    def __init__(self):
        self.t = NoiseTransistor(NoiseSource())
        # Global counter to track total transistor state samples across all operations
        self.total_samples = 0

    def _wait_for_gate_signal(self, expected_conduct):
        """
        Physical synchronization loop. Continually samples the noise transistor 
        until it physically confirms the required state (conduction or isolation).
        """
        while True:
            self.total_samples += 1
            # Sample the actual real-time state of the noise transistor
            current_state = 1 if self.t.conduct() else 0
            if current_state == expected_conduct:
                return current_state

    def NOT(self, a):
        # 0 -> wait until the transistor opens (1)
        # 1 -> wait until the transistor closes (0)
        target = 1 - a
        return self._wait_for_gate_signal(target)

    def AND(self, a, b):
        # Current flows only if both inputs are active AND the transistor conducts
        if a == 1 and b == 1:
            return self._wait_for_gate_signal(1)
        return self._wait_for_gate_signal(0)

    def OR(self, a, b):
        # Current flows if at least one input is active AND the transistor conducts
        if a == 1 or b == 1:
            return self._wait_for_gate_signal(1)
        return self._wait_for_gate_signal(0)

    def XOR(self, a, b):
        # Classic exclusive OR gating driven by the stabilized noise switch
        if a != b:
            return self._wait_for_gate_signal(1)
        return self._wait_for_gate_signal(0)

    def half_adder(self, a, b):
        return self.XOR(a, b), self.AND(a, b)

    def full_adder(self, a, b, cin):
        s1, c1 = self.half_adder(a, b)
        s2, c2 = self.half_adder(s1, cin)
        return s2, self.OR(c1, c2)

    def add(self, x, y, width=8):
        carry, out = 0, 0
        for i in range(width):
            a, b = (x >> i) & 1, (y >> i) & 1
            bit, carry = self.full_adder(a, b, carry)
            out |= bit << i
        return out


if __name__ == "__main__":
    calc = NoiseLogic()
    
    # Perform the calculation
    result = calc.add(13, 29)
    
    # Print the absolute result and stochastic efficiency stats
    print("--- Aethel Stochastic Calculator Execution ---")
    print(f"13 + 29 = {result}")
    print(f"Total transistor samples processed: {calc.total_samples}")
    print("----------------------------------------------")



		 

Physical entropy source based on network timing noise (jitter, queueing artifacts, scheduler drift), with conservative entropy accounting and a cryptographic extractor.

Design principles:
1. Raw network timings are NOT uniform. They are debiased and conditioned.
2. Entropy is accounted conservatively (min-entropy, NIST SP 800-90B style).
3. Network entropy is ALWAYS mixed with os.urandom(), so output quality is
never worse than the system CSPRNG, even under a hostile network.
4. Health tests (repetition count, adaptive proportion) gate the source.

from __future__ import annotations

import hashlib
import hmac
import math
import os
import socket
import statistics
import struct
import time
from collections import Counter
from dataclasses import dataclass, field


# ---------------------------------------------------------------------------
# 1. Raw noise sampling
# ---------------------------------------------------------------------------

@dataclass
class NoiseSource:
    """Samples raw timing noise from one or more network paths.

    Multiple targets are rotated so that a single congested or adversarial
    path cannot dominate the entropy pool.
    """

    targets: tuple = (
        ("1.1.1.1", 80),
        ("8.8.8.8", 53),
        ("9.9.9.9", 53),
        ("208.67.222.222", 53),
    )
    window: int = 256
    timeout: float = 1.0

    history: list = field(default_factory=list)
    _idx: int = 0
    samples_taken: int = 0

    def sample(self) -> int:
        """One TCP-connect attempt; returns RTT in nanoseconds."""
        host, port = self.targets[self._idx % len(self.targets)]
        self._idx += 1

        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        s.settimeout(self.timeout)
        t0 = time.perf_counter_ns()
        try:
            s.connect_ex((host, port))
        except OSError:
            pass
        finally:
            rtt = time.perf_counter_ns() - t0
            s.close()

        self.history.append(rtt)
        if len(self.history) > self.window:
            self.history.pop(0)
        self.samples_taken += 1
        return rtt

    def jitter(self) -> float:
        """Mean absolute successive difference — the classic jitter metric."""
        if len(self.history) < 4:
            return 0.0
        d = [abs(b - a) for a, b in zip(self.history, self.history[1:])]
        return statistics.fmean(d)


# ---------------------------------------------------------------------------
# 2. Bit extraction from raw timings
# ---------------------------------------------------------------------------

class BitExtractor:
    """Turns noisy integers into a debiased bit stream.

    Two stages:
      * LSB slicing  - keep only the low bits of the *difference* between
                       consecutive samples. Deterministic structure (base
                       latency, path length) lives in the high bits; the low
                       bits are dominated by the noise floor.
      * Von Neumann  - consume bit pairs, emit 0 for (0,1) and 1 for (1,0),
                       discard (0,0)/(1,1). Removes static bias at the cost
                       of throughput. Bias-free for i.i.d. input.
    """

    def __init__(self, lsb_bits: int = 8):
        self.lsb_bits = lsb_bits
        self.mask = (1 << lsb_bits) - 1
        self._prev: int | None = None
        self._pair: int | None = None
        self.raw_bits_in = 0
        self.bits_out = 0

    def feed(self, rtt_ns: int) -> list[int]:
        """Feed one raw sample, get zero or more debiased bits back."""
        if self._prev is None:
            self._prev = rtt_ns
            return []

        delta = abs(rtt_ns - self._prev)
        self._prev = rtt_ns

        out: list[int] = []
        for i in range(self.lsb_bits):
            bit = (delta >> i) & 1
            self.raw_bits_in += 1
            if self._pair is None:
                self._pair = bit
            else:
                a, b = self._pair, bit
                self._pair = None
                if a != b:
                    out.append(a)          # von Neumann
                    self.bits_out += 1
        return out

    @property
    def efficiency(self) -> float:
        if not self.raw_bits_in:
            return 0.0
        return self.bits_out / self.raw_bits_in


# ---------------------------------------------------------------------------
# 3. Health tests (NIST SP 800-90B continuous tests)
# ---------------------------------------------------------------------------

class HealthTests:
    """Detects catastrophic source failure — stuck source, hard bias."""

    def __init__(self, rep_cutoff: int = 32, prop_window: int = 512,
                 prop_cutoff: float = 0.65):
        self.rep_cutoff = rep_cutoff
        self.prop_window = prop_window
        self.prop_cutoff = prop_cutoff
        self._last: int | None = None
        self._run = 0
        self._buf: list[int] = []
        self.failures = 0

    def update(self, bit: int) -> bool:
        """Returns True if the source currently looks healthy."""
        ok = True

        # Repetition count test: identical value repeated too many times.
        if bit == self._last:
            self._run += 1
            if self._run >= self.rep_cutoff:
                ok = False
        else:
            self._last, self._run = bit, 1

        # Adaptive proportion test: one symbol dominating the window.
        self._buf.append(bit)
        if len(self._buf) >= self.prop_window:
            ones = sum(self._buf)
            frac = max(ones, len(self._buf) - ones) / len(self._buf)
            if frac > self.prop_cutoff:
                ok = False
            self._buf.clear()

        if not ok:
            self.failures += 1
        return ok


# ---------------------------------------------------------------------------
# 4. Conservative entropy estimation
# ---------------------------------------------------------------------------

def min_entropy_per_bit(bits: list[int]) -> float:
    """Min-entropy of the empirical distribution: -log2(max p_i).

    Deliberately pessimistic. Reports 0.0 for short samples so the pool
    never over-credits itself.
    """
    n = len(bits)
    if n < 128:
        return 0.0
    c = Counter(bits)
    p_max = max(c.values()) / n
    return -math.log2(p_max)


def markov_min_entropy(bits: list[int]) -> float:
    """First-order Markov min-entropy estimate.

    Network noise is strongly autocorrelated (RTT_t depends on RTT_{t-1}).
    An IID estimate would flatter the source; this penalises that structure.
    """
    if len(bits) < 256:
        return 0.0
    trans: Counter = Counter(zip(bits, bits[1:]))
    worst = 0.0
    for state in (0, 1):
        total = trans[(state, 0)] + trans[(state, 1)]
        if total == 0:
            continue
        p = max(trans[(state, 0)], trans[(state, 1)]) / total
        worst = max(worst, -math.log2(p) if p < 1 else 0.0)
    return worst


# ---------------------------------------------------------------------------
# 5. Entropy pool + cryptographic extractor
# ---------------------------------------------------------------------------

class EntropyPool:
    """Accumulates conditioned network entropy, extracts via HMAC-DRBG.

    The pool is *always* seeded from os.urandom() as well. Network noise is
    additional entropy, never the sole source. This is the difference between
    "novel entropy source" and "downgrade attack waiting to happen".
    """

    SEED_LEN = 32
    RESEED_INTERVAL = 1 << 16   # outputs before forced reseed

    def __init__(self, source: NoiseSource | None = None,
                 lsb_bits: int = 8, credit_per_bit: float = 0.25):
        self.src = source or NoiseSource()
        self.extractor = BitExtractor(lsb_bits=lsb_bits)
        self.health = HealthTests()
        self.credit_per_bit = credit_per_bit   # heavily discounted

        self._pool = hashlib.sha512()
        self._pool.update(os.urandom(64))      # never start empty
        self._credited_bits = 0.0
        self._recent: list[int] = []

        self._key = b"\x00" * self.SEED_LEN
        self._V = b"\x01" * self.SEED_LEN
        self._reseed_counter = self.RESEED_INTERVAL  # force reseed on first use
        self.stats: dict = {}

    # -- entropy input ------------------------------------------------------

    def harvest(self, samples: int = 32) -> float:
        """Sample the network, condition the bits, credit min-entropy.

        Returns the number of newly credited entropy bits.
        """
        gained = 0.0
        for _ in range(samples):
            rtt = self.src.sample()
            for bit in self.extractor.feed(rtt):
                healthy = self.health.update(bit)
                self._pool.update(bytes([bit]))
                self._recent.append(bit)
                if healthy:
                    gained += self.credit_per_bit

        # Also absorb the coarse jitter figure and a high-resolution timestamp.
        self._pool.update(struct.pack(" 4096:
            self._recent = self._recent[-4096:]

        # Cap the credit by the measured min-entropy, whichever is lower.
        if len(self._recent) >= 256:
            measured = min(
                min_entropy_per_bit(self._recent),
                markov_min_entropy(self._recent),
            )
            gained = min(gained, measured * len(self._recent))
            self.stats["min_entropy_per_bit"] = round(measured, 4)

        self._credited_bits += gained
        self.stats.update(
            credited_bits=round(self._credited_bits, 2),
            vn_efficiency=round(self.extractor.efficiency, 4),
            health_failures=self.health.failures,
            samples=self.src.samples_taken,
        )
        return gained

    # -- HMAC-DRBG (NIST SP 800-90A) ---------------------------------------

    def _hmac(self, key: bytes, data: bytes) -> bytes:
        return hmac.new(key, data, hashlib.sha256).digest()

    def _update(self, provided: bytes = b"") -> None:
        self._key = self._hmac(self._key, self._V + b"\x00" + provided)
        self._V = self._hmac(self._key, self._V)
        if provided:
            self._key = self._hmac(self._key, self._V + b"\x01" + provided)
            self._V = self._hmac(self._key, self._V)

    def reseed(self, min_entropy_bits: float = 128.0) -> None:
        """Reseed from (network pool || os.urandom).

        If the network hasn't supplied enough credited entropy, we harvest
        more — but we never block forever: os.urandom carries the floor.
        """
        attempts = 0
        while self._credited_bits < min_entropy_bits and attempts < 8:
            self.harvest(samples=24)
            attempts += 1

        network_seed = self._pool.digest()
        system_seed = os.urandom(48)
        self._update(network_seed + system_seed)

        # Re-key the pool without discarding its accumulated state.
        self._pool = hashlib.sha512(network_seed + os.urandom(32))
        self._credited_bits = 0.0
        self._reseed_counter = 0

    def _generate(self, nbytes: int) -> bytes:
        if self._reseed_counter >= self.RESEED_INTERVAL:
            self.reseed()

        out = b""
        while len(out) < nbytes:
            self._V = self._hmac(self._key, self._V)
            out += self._V
        self._update()
        self._reseed_counter += 1
        return out[:nbytes]


# ---------------------------------------------------------------------------
# 6. Public API
# ---------------------------------------------------------------------------

class AethelRandom:
    """Random number generator seeded by network noise.

    API mirrors the standard library where it makes sense.

    NOT a drop-in replacement for `secrets` in adversarial settings unless
    you have audited the entropy path. Use `secrets` when in doubt.
    """

    def __init__(self, pool: EntropyPool | None = None, warmup: int = 64):
        self.pool = pool or EntropyPool()
        if warmup:
            self.pool.harvest(samples=warmup)
        self.pool.reseed()

    # -- primitives --------------------------------------------------------

    def random_bytes(self, n: int) -> bytes:
        return self.pool._generate(n)

    def random_bits(self, k: int) -> int:
        if k <= 0:
            return 0
        nbytes = (k + 7) // 8
        val = int.from_bytes(self.random_bytes(nbytes), "big")
        return val >> (nbytes * 8 - k)

    def random(self) -> float:
        """Uniform float in [0, 1) with full 53-bit mantissa precision."""
        return self.random_bits(53) / (1 << 53)

    def randbelow(self, n: int) -> int:
        """Uniform int in [0, n) — rejection sampling, no modulo bias."""
        if n <= 0:
            raise ValueError("n must be positive")
        k = n.bit_length()
        while True:
            v = self.random_bits(k)
            if v < n:
                return v

    def randint(self, a: int, b: int) -> int:
        """Uniform int in [a, b], inclusive."""
        if b < a:
            raise ValueError("empty range")
        return a + self.randbelow(b - a + 1)

    def choice(self, seq):
        if not seq:
            raise IndexError("cannot choose from an empty sequence")
        return seq[self.randbelow(len(seq))]

    def shuffle(self, seq: list) -> None:
        """In-place Fisher-Yates."""
        for i in range(len(seq) - 1, 0, -1):
            j = self.randbelow(i + 1)
            seq[i], seq[j] = seq[j], seq[i]

    def sample(self, population, k: int) -> list:
        if k > len(population):
            raise ValueError("sample larger than population")
        pool = list(population)
        out = []
        for i in range(k):
            j = i + self.randbelow(len(pool) - i)
            pool[i], pool[j] = pool[j], pool[i]
            out.append(pool[i])
        return out

    # -- introspection -----------------------------------------------------

    def report(self) -> dict:
        """Honest status of the entropy path."""
        s = dict(self.pool.stats)
        s["jitter_ns"] = round(self.pool.src.jitter(), 1)
        s["source_healthy"] = self.pool.health.failures == 0
        s["mixed_with_os_urandom"] = True
        return s


# ---------------------------------------------------------------------------

if __name__ == "__main__":
    rng = AethelRandom(warmup=48)

    print("bytes  :", rng.random_bytes(16).hex())
    print("float  :", rng.random())
    print("d20    :", [rng.randint(1, 20) for _ in range(10)])

    deck = list(range(10))
    rng.shuffle(deck)
    print("shuffle:", deck)

    print("\nentropy path report:")
    for k, v in sorted(rng.report().items()):
        print(f"  {k:24s} {v}")


		 

The idea in its bluntest form. Both peers observe the same shared statistical environment and derive role, timing and symbol from agreed patterns. Nothing is transmitted; both sides only listen and act. Published as a concept for critique.

ALGORITHM AethelNoiseProtocol(shared_seed, shared_observation_plan)

  loop forever:

      # 1. both peers observe the identical noise environment
      W  = observe_noise_window(shared_observation_plan)
      F  = extract_features(W)                 # entropy, jitter, drift, phase

      # 2. derive a common frame from the noise itself
      frame_id = hash(shared_seed, quantize(F.slow_components))

      # 3. role assignment from a deterministic function of the frame
      role = SENDER if parity(frame_id) == 0 else RECEIVER

      # 4. symbol channel: the sender does not transmit,
      #    it chooses WHICH pattern in the ambient noise to act upon
      if role == SENDER:
          target_pattern = select_pattern(message_bit, F)
          wait_until_pattern_occurs(target_pattern)   # no packet is emitted

      else:  # RECEIVER
          seen = wait_and_classify_pattern(F, timeout=frame_len)
          bit  = decode(seen)
          append_to_message(bit)

      advance_frame()

  # Open problems, stated honestly:
  #   - do two distant observers share enough correlated noise?
  #   - what is the achievable bitrate, if any? (bits per minute, at best)
  #   - is "acting on" a pattern distinguishable from not acting on it?
  # Phase 3 exists to answer these, and to publish a negative result if that is the answer.