DHCP 只把地址租给你

"""Simulate a DHCP lease lifecycle on a virtual clock (stdlib only)."""
from __future__ import annotations

LEASE_TIME = 100
T1 = LEASE_TIME // 2          # 50: unicast renewal (RFC 2131 default)
T2 = int(LEASE_TIME * 0.875)  # 87: broadcast rebinding (RFC 2131 default)


class DHCPServer:
    def __init__(self, pool):
        self.pool = list(pool)
        self.leases: dict[str, dict] = {}  # ip -> {"client": id, "expiry": int}
        self.now = 0

    def _free_ip(self):
        for ip in self.pool:
            lease = self.leases.get(ip)
            if lease is None or lease["expiry"] <= self.now:
                return ip
        return None

    def discover_offer(self, client_id):
        """DISCOVER -> OFFER (no state change)."""
        ip = self._free_ip()
        assert ip is not None, f"t={self.now}: pool exhausted, no OFFER for {client_id}"
        return ip

    def request_ack(self, client_id, ip):
        """REQUEST -> ACK and bind the lease."""
        assert ip in self.pool, "requested address outside pool"
        lease = self.leases.get(ip)
        assert lease is None or lease["expiry"] <= self.now or lease["client"] == client_id, \
            f"t={self.now}: {ip} held by another client"
        self.leases[ip] = {"client": client_id, "expiry": self.now + LEASE_TIME}
        return {"ip": ip, "lease_time": LEASE_TIME, "t1": T1, "t2": T2,
                "expiry": self.now + LEASE_TIME}

    def renew(self, client_id, ip):
        """Extend a current lease; transport and client states are not modeled."""
        lease = self.leases.get(ip)
        if lease is None or lease["client"] != client_id or lease["expiry"] <= self.now:
            return False  # expired: server already reclaimed
        lease["expiry"] = self.now + LEASE_TIME
        return True

    def tick(self, dt):
        self.now += dt
        # Reclaim expired leases so a new client can reuse them.
        for ip, lease in list(self.leases.items()):
            if lease["expiry"] <= self.now:
                del self.leases[ip]


def check(cond, msg):
    assert cond, msg


def main():
    server = DHCPServer(["192.168.1.10", "192.168.1.11"])
    # 1. DORA: client c1 joins.
    offered = server.discover_offer("c1")
    check(offered == "192.168.1.10", f"expected first free IP, got {offered}")
    ack = server.request_ack("c1", offered)
    check((ack["lease_time"], ack["t1"], ack["t2"]) == (100, 50, 87), f"bad timers: {ack}")
    check(ack["expiry"] == 100, f"expiry should be 0+100, got {ack['expiry']}")
    print(f"t={server.now}: c1 ACK {ack['ip']} lease=100 T1=50 T2=87 expiry=100")
    # 2. T1 unicast renewal at half lease.
    server.tick(T1)  # now=50
    check(server.renew("c1", offered), "T1 renewal should be ACKed")
    check(server.leases[offered]["expiry"] == 150, "T1 must extend expiry to 50+100")
    print(f"t={server.now}: c1 T1 unicast renew OK expiry=150")
    # 3. Miss T1, rebind at T2 via broadcast.
    server.tick(T1)       # now=100: renewed T1; pretend this attempt got no ACK
    server.tick(T2 - T1)  # now=137: renewed T2, before expiry 150
    check(server.now == 50 + T2, "rebind must use renewed lease origin")
    check(server.renew("c1", offered), "T2 rebinding should be ACKed")
    check(server.leases[offered]["expiry"] == 237, "T2 must extend expiry to 137+100")
    print(f"t={server.now}: c1 T2 broadcast rebind OK expiry=237")
    # 4. Leave without renewing: expiry reclaims the address.
    server.tick(200)  # now=337 > expiry 237
    check(offered not in server.leases, "expired lease must be reclaimed")
    offered2 = server.discover_offer("c2")
    check(offered2 == offered, f"reclaimed {offered} should be reused, got {offered2}")
    ack2 = server.request_ack("c2", offered2)
    check(ack2["expiry"] == server.now + LEASE_TIME, "new lease expiry must restart")
    # Stale renewal from the departed client must fail.
    check(not server.renew("c1", offered), "stale renewal after expiry must not be ACKed")
    print(f"t={server.now}: c1 expired, c2 ACK {ack2['ip']} expiry={ack2['expiry']}")
    print("PASS: discover/offer/request/ack, T1 renew, T2 rebind, expiry reclaim")


if __name__ == "__main__":
    main()