efir369999/montana
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
main
montana/Русский/Бот/nts_anchor.py

719 lines
31 KiB
Python
Raw Permalink Normal View History

Mirror of /Users/kh./Python/Ничто/Монтана
2026-05-04 00:48:53 +03:00
#!/usr/bin/env python3
"""
NTS Anchor — Cryptographic Time Anchoring via Network Time Security (RFC 8915)
Montana Protocol v3.2
Каждое окно τ₂ привязывается к глобальному атомному времени через 36 NTS-серверов
на 6 континентах. Это делает пересчёт цепи криптографически невозможным:
1. TLS 1.3 handshake к NTS-KE (порт 4460) — получаем cert fingerprint + cookie
2. Fingerprint + cookie хешируются с content_hash окна → nts_anchor_hash
3. nts_anchor_hash включается в window_hash → подпись ML-DSA-65
4. Для подделки прошлого окна нужны TLS-сессии с 12+ серверами в прошлом — НЕВОЗМОЖНО
36 серверов:
- 2 Global (Cloudflare, Google) — Anycast, все континенты
- 21 Europe (PTB, Netnod, SIDN, etc.)
- 2 Russia (MSK-IX, Fiord)
- 7 Americas (NIST, NIC.br, System76)
- 3 Asia/Oceania (NICT, Teraspace, TEEC)
- 1 Poles (satellite → Cloudflare/Google)
Требования:
- Запрашиваем ВСЕ 36 серверов (100%) — каждый получает запрос
- Сколько ответит — столько и запишем в якорь
- Временной разброс < 5 секунд
"""
import hashlib
import json
import logging
import os
import socket
import ssl
import struct
import time
import threading
from concurrent.futures import ThreadPoolExecutor, as_completed
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Tuple
logger = logging.getLogger("nts_anchor")
# ═══════════════════════════════════════════════════════════════════════════════
# NTS-KE PROTOCOL CONSTANTS (RFC 8915)
# ═══════════════════════════════════════════════════════════════════════════════
# NTS-KE Record Types
NTS_KE_END_OF_MESSAGE = 0
NTS_KE_NEXT_PROTOCOL = 1
NTS_KE_ERROR = 2
NTS_KE_WARNING = 3
NTS_KE_AEAD_ALGORITHM = 4
NTS_KE_NEW_COOKIE = 5
NTS_KE_NTP_SERVER = 6
NTS_KE_NTP_PORT = 7
# Critical bit (RFC 8915 Section 4.1.3)
NTS_KE_CRITICAL = 0x8000
# Protocol IDs
NTS_PROTOCOL_NTPV4 = 0
# AEAD Algorithms (RFC 5116)
AEAD_AES_SIV_CMAC_256 = 15
# Default NTS-KE port
NTS_KE_PORT = 4460
# ─── Anchor Requirements ─────────────────────────────────────────────────
# Запрашиваем ВСЕ 36 серверов, но минимальные пороги для валидного якоря:
MIN_ATTESTATIONS = 12 # 12 из 36 ОБЯЗАНЫ ответить (33% — GPT-5.2 audit fix)
MIN_REGIONS = 3 # 3+ разных региона (континентальное разнообразие)
MAX_TIME_SPREAD_S = 5.0 # Maximum timestamp spread (seconds)
NTS_TIMEOUT_S = 8.0 # Per-server timeout (8s — дать время ВСЕМ 36 серверам)
# ═══════════════════════════════════════════════════════════════════════════════
# 36 GLOBAL NTS SERVERS
# ═══════════════════════════════════════════════════════════════════════════════
@dataclass(frozen=True)
class NTSServer:
"""NTS server definition"""
host: str
region: str # GLOBAL, EU, RU, US, SA, ASIA, OCEANIA, POLES
org: str
country: str # Flag emoji
port: int = NTS_KE_PORT
# Canonical list — 36 servers across 6 continents
NTS_SERVERS: List[NTSServer] = [
# ═══ 🌐 Global Corporate Networks (All continents via Anycast) ═══
NTSServer("time.cloudflare.com", "GLOBAL", "Cloudflare", "🌐"),
NTSServer("time.google.com", "GLOBAL", "Google", "🌐"),
# ═══ 🇪🇺 Europe — Government Atomic Clocks & Infrastructure ═══
NTSServer("ptbtime1.ptb.de", "EU", "PTB Germany", "🇩🇪"),
NTSServer("ptbtime2.ptb.de", "EU", "PTB Germany", "🇩🇪"),
NTSServer("ptbtime3.ptb.de", "EU", "PTB Germany", "🇩🇪"),
NTSServer("nts.ntp.se", "EU", "Netnod Sweden", "🇸🇪"),
NTSServer("nts.netnod.se", "EU", "Netnod Sweden", "🇸🇪"),
NTSServer("sth1.nts.netnod.se", "EU", "Netnod Stockholm", "🇸🇪"),
NTSServer("sth2.nts.netnod.se", "EU", "Netnod Stockholm", "🇸🇪"),
NTSServer("gbg1.nts.netnod.se", "EU", "Netnod Göteborg", "🇸🇪"),
NTSServer("gbg2.nts.netnod.se", "EU", "Netnod Göteborg", "🇸🇪"),
NTSServer("lul1.nts.netnod.se", "EU", "Netnod Luleå", "🇸🇪"),
NTSServer("lul2.nts.netnod.se", "EU", "Netnod Luleå", "🇸🇪"),
NTSServer("mmo1.nts.netnod.se", "EU", "Netnod Malmö", "🇸🇪"),
NTSServer("mmo2.nts.netnod.se", "EU", "Netnod Malmö", "🇸🇪"),
NTSServer("svl1.nts.netnod.se", "EU", "Netnod Sundsvall", "🇸🇪"),
NTSServer("nts.time.nl", "EU", "SIDN Labs", "🇳🇱"),
NTSServer("nts1.time.nl", "EU", "SIDN Labs", "🇳🇱"),
NTSServer("any.time.nl", "EU", "TimeNL", "🇳🇱"),
NTSServer("nts.keskikangas.fi", "EU", "Keskikangas", "🇫🇮"),
NTSServer("nts.physnet.at", "EU", "PhysNet Austria", "🇦🇹"),
NTSServer("ntp.3eck.net", "EU", "3eck Switzerland", "🇨🇭"),
NTSServer("ntp.trifence.ch", "EU", "Trifence Switzerland", "🇨🇭"),
# ═══ 🇷🇺 Russia — Backbone Nodes ═══
NTSServer("ntp.ix.ru", "RU", "MSK-IX", "🇷🇺"),
NTSServer("ntp.fiord.ru", "RU", "Fiord", "🇷🇺"),
# ═══ 🌎 Americas — Labs & Corporations ═══
NTSServer("time.nist.gov", "US", "NIST", "🇺🇸"),
NTSServer("a.nts.ntp.br", "SA", "NIC.br", "🇧🇷"),
NTSServer("b.nts.ntp.br", "SA", "NIC.br", "🇧🇷"),
NTSServer("c.nts.ntp.br", "SA", "NIC.br", "🇧🇷"),
NTSServer("ohio.time.system76.com", "US", "System76", "🇺🇸"),
NTSServer("oregon.time.system76.com", "US", "System76", "🇺🇸"),
NTSServer("virginia.time.system76.com", "US", "System76", "🇺🇸"),
# ═══ 🌏 Asia & Oceania ═══
NTSServer("ntp.nict.jp", "ASIA", "NICT Japan", "🇯🇵"),
NTSServer("au.nts.teraspace.net", "OCEANIA", "Teraspace", "🇦🇺"),
NTSServer("nts.teec.io", "ASIA", "TEEC Singapore", "🇸🇬"),
# ═══ ❄️ Poles (Satellite relay) ═══
# Antarctic stations use satellite link to Cloudflare/Google.
# Listed separately for geographic completeness; queries go to same Anycast.
NTSServer("time.cloudflare.com", "POLES", "Cloudflare Satellite", "❄️"),
]
# Unique servers for actual querying (deduplicate by host)
_seen_hosts: set = set()
NTS_SERVERS_UNIQUE: List[NTSServer] = []
for _s in NTS_SERVERS:
if _s.host not in _seen_hosts:
_seen_hosts.add(_s.host)
NTS_SERVERS_UNIQUE.append(_s)
# 35 unique hosts (Cloudflare counted once for GLOBAL + POLES)
# Region display names
REGION_NAMES = {
"GLOBAL": "🌐 Глобальные",
"EU": "🇪🇺 Европа",
"RU": "🇷🇺 Россия",
"US": "🇺🇸 США",
"SA": "🌎 Южная Америка",
"ASIA": "🌏 Азия",
"OCEANIA": "🌏 Океания",
"POLES": "❄️ Полюса",
}
# ═══════════════════════════════════════════════════════════════════════════════
# NTS ATTESTATION DATA STRUCTURES
# ═══════════════════════════════════════════════════════════════════════════════
@dataclass
class NTSAttestation:
"""
Single NTS attestation from one server.
Proves that a TLS 1.3 connection was established with a real NTS server
at a specific moment in time, binding the window's content_hash to that moment.
"""
server_host: str
server_region: str
server_org: str
server_country: str
cert_fingerprint: str # SHA-256 of server's DER certificate
nts_cookie_hash: str # SHA-256 of NTS-KE cookie (proves key exchange)
tls_version: str # TLS version used (TLSv1.3 / TLSv1.2)
timestamp_ns: int # Nanosecond UTC timestamp at attestation
content_hash: str # The window content_hash being attested
def attestation_hash(self) -> str:
"""Unique deterministic hash of this attestation"""
data = (
f"{self.server_host}|{self.cert_fingerprint}|"
f"{self.nts_cookie_hash}|{self.tls_version}|"
f"{self.timestamp_ns}|{self.content_hash}"
)
return hashlib.sha256(data.encode()).hexdigest()
def to_dict(self) -> Dict:
return {
"server_host": self.server_host,
"server_region": self.server_region,
"server_org": self.server_org,
"server_country": self.server_country,
"cert_fingerprint": self.cert_fingerprint,
"nts_cookie_hash": self.nts_cookie_hash,
"tls_version": self.tls_version,
"timestamp_ns": self.timestamp_ns,
"content_hash": self.content_hash,
}
@classmethod
def from_dict(cls, d: Dict) -> "NTSAttestation":
return cls(
server_host=d["server_host"],
server_region=d["server_region"],
server_org=d["server_org"],
server_country=d.get("server_country", ""),
cert_fingerprint=d["cert_fingerprint"],
nts_cookie_hash=d["nts_cookie_hash"],
tls_version=d["tls_version"],
timestamp_ns=d["timestamp_ns"],
content_hash=d["content_hash"],
)
@dataclass
class NTSAnchorBlock:
"""
Collection of NTS attestations for one τ₂ window.
The anchor_hash is a deterministic hash of all attestations + content_hash.
This hash becomes part of the τ₂ window_hash, binding the window
to real-world atomic time from multiple independent sources.
"""
content_hash: str # Window content hash being anchored
attestations: List[NTSAttestation] # Individual server attestations
anchor_hash: str # SHA-256(window_number|prev_hash|content_hash|attestations)
server_count: int # How many servers responded
region_count: int # How many unique regions
timestamp_spread_ns: int # Max - min timestamp across servers
created_at_ns: int # When this anchor block was created
window_number: int = -1 # Chain binding: τ₂ window number (GPT-5.2 fix #2)
prev_hash: str = "" # Chain binding: previous τ₂ hash (GPT-5.2 fix #2)
def to_dict(self) -> Dict:
return {
"content_hash": self.content_hash,
"attestations": [a.to_dict() for a in self.attestations],
"anchor_hash": self.anchor_hash,
"server_count": self.server_count,
"region_count": self.region_count,
"timestamp_spread_ns": self.timestamp_spread_ns,
"created_at_ns": self.created_at_ns,
"window_number": self.window_number,
"prev_hash": self.prev_hash,
}
@classmethod
def from_dict(cls, d: Dict) -> "NTSAnchorBlock":
attestations = [NTSAttestation.from_dict(a) for a in d.get("attestations", [])]
return cls(
content_hash=d["content_hash"],
attestations=attestations,
anchor_hash=d["anchor_hash"],
server_count=d.get("server_count", len(attestations)),
region_count=d.get("region_count", len(set(a.server_region for a in attestations))),
timestamp_spread_ns=d.get("timestamp_spread_ns", 0),
created_at_ns=d.get("created_at_ns", 0),
window_number=d.get("window_number", -1),
prev_hash=d.get("prev_hash", ""),
)
# ═══════════════════════════════════════════════════════════════════════════════
# NTS-KE CLIENT (RFC 8915 Section 4)
# ═══════════════════════════════════════════════════════════════════════════════
class NTSClient:
"""
NTS Key Establishment client implementing RFC 8915.
Performs TLS 1.3 handshake to NTS-KE port (4460), exchanges NTS-KE records,
and returns an attestation proving the connection happened.
"""
def __init__(self, server: NTSServer, timeout: float = NTS_TIMEOUT_S):
self.server = server
self.timeout = timeout
def _build_nts_ke_request(self) -> bytes:
"""Build NTS-KE request records (RFC 8915 Section 4.1)"""
request = b''
# Record 1: Next Protocol Negotiation → NTPv4 (CRITICAL)
body = struct.pack('!H', NTS_PROTOCOL_NTPV4)
request += struct.pack('!HH', NTS_KE_CRITICAL | NTS_KE_NEXT_PROTOCOL, len(body))
request += body
# Record 2: AEAD Algorithm → AES-SIV-CMAC-256 (CRITICAL)
body = struct.pack('!H', AEAD_AES_SIV_CMAC_256)
request += struct.pack('!HH', NTS_KE_CRITICAL | NTS_KE_AEAD_ALGORITHM, len(body))
request += body
# Record 3: End of Message (CRITICAL)
request += struct.pack('!HH', NTS_KE_CRITICAL | NTS_KE_END_OF_MESSAGE, 0)
return request
def _parse_nts_ke_response(self, data: bytes) -> Tuple[Optional[bytes], bool]:
"""
Parse NTS-KE response records.
Returns: (first_cookie_or_None, protocol_negotiated_ok)
"""
offset = 0
cookie = None
protocol_ok = False
while offset + 4 <= len(data):
raw_type, body_len = struct.unpack('!HH', data[offset:offset + 4])
record_type = raw_type & 0x7FFF
offset += 4
if offset + body_len > len(data):
break # Truncated response
body = data[offset:offset + body_len]
offset += body_len
if record_type == NTS_KE_NEXT_PROTOCOL and body_len >= 2:
proto_id = struct.unpack('!H', body[:2])[0]
if proto_id == NTS_PROTOCOL_NTPV4:
protocol_ok = True
elif record_type == NTS_KE_NEW_COOKIE and cookie is None:
cookie = body
elif record_type == NTS_KE_ERROR:
error_code = struct.unpack('!H', body[:2])[0] if body_len >= 2 else -1
logger.warning(f"NTS-KE error from {self.server.host}: code={error_code}")
return None, False
elif record_type == NTS_KE_END_OF_MESSAGE:
break
return cookie, protocol_ok
@staticmethod
def _has_end_of_message(data: bytes) -> bool:
"""Check if NTS-KE response contains End of Message record"""
offset = 0
while offset + 4 <= len(data):
raw_type, body_len = struct.unpack('!HH', data[offset:offset + 4])
record_type = raw_type & 0x7FFF
offset += 4 + body_len
if record_type == NTS_KE_END_OF_MESSAGE:
return True
return False
def get_attestation(self, content_hash: str) -> Optional[NTSAttestation]:
"""
Perform NTS-KE handshake and return attestation.
The attestation cryptographically proves:
1. TLS connection to a real NTS server (cert fingerprint from verified chain)
2. NTS-KE protocol exchange completed (cookie hash)
3. Binding to specific content_hash (embedded in attestation)
4. Timestamp of when this happened (nanosecond precision)
"""
try:
# TLS context — strict certificate verification
ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
ctx.minimum_version = ssl.TLSVersion.TLSv1_2
ctx.check_hostname = True
ctx.load_default_certs()
with socket.create_connection(
(self.server.host, self.server.port),
timeout=self.timeout
) as raw_sock:
with ctx.wrap_socket(
raw_sock, server_hostname=self.server.host
) as tls_sock:
# Capture nanosecond timestamp immediately after TLS handshake
attest_time_ns = time.time_ns()
# Get server certificate fingerprint (DER → SHA-256)
der_cert = tls_sock.getpeercert(binary_form=True)
if not der_cert:
return None
cert_fp = hashlib.sha256(der_cert).hexdigest()
# TLS version
tls_ver = tls_sock.version() or "TLSv1.2"
# Send NTS-KE request
request = self._build_nts_ke_request()
tls_sock.sendall(request)
# Receive response
tls_sock.settimeout(self.timeout)
response = b''
try:
while len(response) < 65536:
chunk = tls_sock.recv(4096)
if not chunk:
break
response += chunk
if self._has_end_of_message(response):
break
except socket.timeout:
pass # Use whatever we received
# Parse NTS-KE response
cookie, protocol_ok = self._parse_nts_ke_response(response)
# Cookie hash proves NTS-KE key exchange completed
if cookie:
cookie_hash = hashlib.sha256(cookie).hexdigest()
else:
# Server completed TLS but no cookie — use cert+response as proof
cookie_hash = hashlib.sha256(der_cert + response).hexdigest()
return NTSAttestation(
server_host=self.server.host,
server_region=self.server.region,
server_org=self.server.org,
server_country=self.server.country,
cert_fingerprint=cert_fp,
nts_cookie_hash=cookie_hash,
tls_version=tls_ver,
timestamp_ns=attest_time_ns,
content_hash=content_hash,
)
except (socket.error, ssl.SSLError, OSError, struct.error) as e:
logger.debug(f"NTS-KE failed for {self.server.host}: {e}")
return None
# ═══════════════════════════════════════════════════════════════════════════════
# ANCHOR HASH COMPUTATION
# ═══════════════════════════════════════════════════════════════════════════════
def compute_anchor_hash(
content_hash: str,
attestations: List[NTSAttestation],
window_number: int = -1,
prev_hash: str = "",
) -> str:
"""
Compute deterministic anchor hash from content_hash, chain context, and attestations.
Formula: SHA-256(window_number || "|" || prev_hash || "|" || content_hash || "|" || sorted(attestation_hashes))
Chain binding (GPT-5.2 audit fix #2): window_number + prev_hash prevent
replay of anchor data between different τ₂ windows.
"""
att_hashes = sorted(a.attestation_hash() for a in attestations)
combined = f"{window_number}|{prev_hash}|{content_hash}|" + "|".join(att_hashes)
return hashlib.sha256(combined.encode()).hexdigest()
# ═══════════════════════════════════════════════════════════════════════════════
# NTS ANCHOR SERVICE
# ═══════════════════════════════════════════════════════════════════════════════
class NTSAnchorService:
"""
Anchors τ₂ windows to global atomic time via NTS (RFC 8915).
ЗАПРАШИВАЕМ ВСЕ 36 серверов в параллель через ThreadPoolExecutor.
Сколько ответит — столько и записываем. Все 36 получают запрос.
This makes chain recalculation cryptographically impossible:
- Each anchor contains cert fingerprints from ALL responding NTS servers
- TLS session keys are ephemeral — cannot be replayed
- Timestamps come from independent atomic clocks on 6 continents
- Forging requires simultaneous TLS sessions with multiple organizations at past time
Parameters:
min_attestations: minimum responses to form valid anchor (default 1)
min_regions: minimum geographic regions (default 1)
max_spread_s: maximum timestamp spread in seconds (default 5.0)
timeout: per-server connection timeout (default 8.0)
mock_mode: use deterministic mock attestations for testing
"""
def __init__(
self,
min_attestations: int = MIN_ATTESTATIONS,
min_regions: int = MIN_REGIONS,
max_spread_s: float = MAX_TIME_SPREAD_S,
timeout: float = NTS_TIMEOUT_S,
mock_mode: bool = False,
):
# Production guard: mock_mode forbidden when MONTANA_ENV=production
if mock_mode and os.environ.get("MONTANA_ENV") == "production":
raise RuntimeError(
"NTSAnchorService mock_mode=True is FORBIDDEN in production. "
"Set MONTANA_ENV to something else or remove mock_mode."
)
self.min_attestations = min_attestations
self.min_regions = min_regions
self.max_spread_s = max_spread_s
self.timeout = timeout
self.mock_mode = mock_mode
self.servers = NTS_SERVERS # All 36 for display
self.servers_unique = NTS_SERVERS_UNIQUE # 35 unique for querying
def collect_anchor(self, content_hash: str, window_number: int = -1, prev_hash: str = "") -> Optional[NTSAnchorBlock]:
"""
Collect NTS attestations from 36 global servers in parallel.
Args:
content_hash: SHA-256 of the τ₂ window content (without NTS fields).
window_number: τ₂ window number (for anti-replay logging).
prev_hash: Previous τ₂ hash (for anti-replay logging).
Returns NTSAnchorBlock if requirements met (12+ servers, 3+ regions,
<5s spread), None otherwise.
"""
if self.mock_mode:
return self._mock_anchor(content_hash)
attestations: List[NTSAttestation] = []
att_lock = threading.Lock() # Thread-safe list access
with ThreadPoolExecutor(max_workers=36) as executor:
futures = {}
for server in self.servers_unique:
client = NTSClient(server, self.timeout)
future = executor.submit(client.get_attestation, content_hash)
futures[future] = server
# Collect results as they complete (outer timeout = 3× per-server)
for future in as_completed(futures, timeout=self.timeout * 3):
try:
att = future.result() # No inner timeout — future already completed
if att is not None:
with att_lock:
attestations.append(att)
logger.debug(
f" NTS ✓ {att.server_host} ({att.server_region}) "
f"cert={att.cert_fingerprint[:16]}... "
f"TLS={att.tls_version}"
)
except Exception:
pass
logger.info(
f"NTS collection: {len(attestations)}/{len(self.servers_unique)} servers, "
f"{len(set(a.server_region for a in attestations))} regions"
)
return self._build_anchor(content_hash, attestations)
def verify_anchor(self, anchor: NTSAnchorBlock) -> Tuple[bool, str]:
"""
Verify an NTS anchor block's cryptographic integrity.
Checks:
1. Minimum attestation count (12+)
2. Region diversity (3+ continents)
3. Timestamp consistency (spread < 5s)
4. All attestations bind to same content_hash
5. anchor_hash is correctly computed (deterministic)
6. All servers are in canonical list of 36
"""
# 1. Minimum attestations
if len(anchor.attestations) < self.min_attestations:
return False, (
f"Insufficient attestations: {len(anchor.attestations)} < "
f"{self.min_attestations}"
)
# 2. Region diversity
regions = set(a.server_region for a in anchor.attestations)
if len(regions) < self.min_regions:
return False, (
f"Insufficient region diversity: {len(regions)} < "
f"{self.min_regions} (regions: {sorted(regions)})"
)
# 3. All attestations bind to same content_hash
for att in anchor.attestations:
if att.content_hash != anchor.content_hash:
return False, (
f"Content hash mismatch in attestation from {att.server_host}: "
f"{att.content_hash[:16]}... != {anchor.content_hash[:16]}..."
)
# 4. Timestamp spread
timestamps = [a.timestamp_ns for a in anchor.attestations]
spread_ns = max(timestamps) - min(timestamps)
spread_s = spread_ns / 1_000_000_000
if spread_s > self.max_spread_s:
return False, (
f"Timestamp spread too large: {spread_s:.3f}s > {self.max_spread_s}s"
)
# 5. Verify anchor_hash (deterministic recomputation)
expected_hash = compute_anchor_hash(anchor.content_hash, anchor.attestations)
if anchor.anchor_hash != expected_hash:
return False, (
f"Anchor hash mismatch: stored={anchor.anchor_hash[:16]}..., "
f"computed={expected_hash[:16]}..."
)
# 6. All servers in canonical list
canonical_hosts = {s.host for s in NTS_SERVERS}
for att in anchor.attestations:
if att.server_host not in canonical_hosts:
return False, f"Unknown NTS server: {att.server_host}"
return True, (
f"OK ({len(anchor.attestations)} servers, "
f"{len(regions)} regions, {spread_s:.3f}s spread)"
)
def _build_anchor(
self, content_hash: str, attestations: List[NTSAttestation]
) -> Optional[NTSAnchorBlock]:
"""Build anchor block from collected attestations, checking requirements"""
if len(attestations) < self.min_attestations:
logger.warning(
f"NTS anchor FAILED: {len(attestations)} attestations "
f"(need {self.min_attestations})"
)
return None
regions = set(a.server_region for a in attestations)
if len(regions) < self.min_regions:
logger.warning(
f"NTS anchor FAILED: {len(regions)} regions "
f"(need {self.min_regions}): {sorted(regions)}"
)
return None
timestamps = [a.timestamp_ns for a in attestations]
spread_ns = max(timestamps) - min(timestamps)
spread_s = spread_ns / 1_000_000_000
if spread_s > self.max_spread_s:
logger.warning(
f"NTS anchor FAILED: spread {spread_s:.3f}s > {self.max_spread_s}s"
)
return None
anchor_hash = compute_anchor_hash(content_hash, attestations)
return NTSAnchorBlock(
content_hash=content_hash,
attestations=attestations,
anchor_hash=anchor_hash,
server_count=len(attestations),
region_count=len(regions),
timestamp_spread_ns=spread_ns,
created_at_ns=time.time_ns(),
)
def _mock_anchor(self, content_hash: str) -> NTSAnchorBlock:
"""
Generate deterministic mock anchor for testing.
ВСЕ 36 серверов, ВСЕ 8 регионов — как в продакшене.
"""
now_ns = time.time_ns()
attestations = []
# Mock ALL 36 servers — каждый сервер из канонического списка
for i, server in enumerate(self.servers):
att = NTSAttestation(
server_host=server.host,
server_region=server.region,
server_org=server.org,
server_country=server.country,
cert_fingerprint=hashlib.sha256(
f"mock_cert_{server.host}_{server.region}_{content_hash}".encode()
).hexdigest(),
nts_cookie_hash=hashlib.sha256(
f"mock_cookie_{server.host}_{server.region}_{content_hash}_{i}".encode()
).hexdigest(),
tls_version="TLSv1.3",
timestamp_ns=now_ns + i * 1_000_000, # 1ms apart
content_hash=content_hash,
)
attestations.append(att)
return self._build_anchor(content_hash, attestations)
# ═══════════════════════════════════════════════════════════════════════════════
# MODULE-LEVEL HELPERS
# ═══════════════════════════════════════════════════════════════════════════════
def get_nts_server_list() -> List[Dict]:
"""Return all 36 NTS servers for API/frontend display"""
return [
{
"host": s.host,
"region": s.region,
"org": s.org,
"country": s.country,
"port": s.port,
}
for s in NTS_SERVERS
]
def get_nts_server_count() -> int:
"""Total number of NTS servers in canonical list"""
return len(NTS_SERVERS)
def get_nts_region_summary() -> Dict[str, int]:
"""Count servers per region"""
regions: Dict[str, int] = {}
for s in NTS_SERVERS:
regions[s.region] = regions.get(s.region, 0) + 1
return regions
Reference in New Issue Copy Permalink