Chrome WebRTC 防泄露测试：用本地 STUN 验证网络策略

"""Original loopback-only STUN teaching experiment; not an Internet STUN server."""
import importlib.metadata
import ipaddress
import json
import platform
from pathlib import Path
import socket
import struct
import threading
from playwright.sync_api import sync_playwright

COOKIE=0x2112A442
TEST_IP='203.0.113.42'
SCRIPT='''async (port) => {
  const pc = new RTCPeerConnection({iceServers:[{urls:`stun:127.0.0.1:${port}`} ]});
  const candidates=[];
  try {
    const finished=new Promise(resolve => {
      pc.onicecandidate=e => { if(e.candidate)candidates.push({type:e.candidate.type,address:e.candidate.address}); };
      pc.onicegatheringstatechange=()=> {if(pc.iceGatheringState==='complete')resolve();};
      setTimeout(resolve,4000);
    });
    pc.createDataChannel('test');
    await pc.setLocalDescription(await pc.createOffer());
    await finished;
    return {state:pc.iceGatheringState,types:candidates.map(c=>c.type),hasTestMapping:candidates.some(c=>c.type==='srflx' && c.address==='203.0.113.42')};
  } finally {pc.close();}
}'''


def run():
    stop=threading.Event();count=[0];lock=threading.Lock()
    sock=socket.socket(socket.AF_INET,socket.SOCK_DGRAM);sock.bind(('127.0.0.1',0));sock.settimeout(.1)
    port=sock.getsockname()[1]
    def serve():
        while not stop.is_set():
            try:data,peer=sock.recvfrom(4096)
            except socket.timeout:continue
            if len(data)<20:continue
            kind,length,cookie,transaction=struct.unpack('!HHI12s',data[:20])
            if kind!=1 or cookie!=COOKIE or len(data)<20+length:continue
            with lock:count[0]+=1
            attribute=struct.pack('!HHBBHI',0x20,8,0,1,peer[1]^(COOKIE>>16),int(ipaddress.IPv4Address(TEST_IP))^COOKIE)
            sock.sendto(struct.pack('!HHI12s',0x101,len(attribute),COOKIE,transaction)+attribute,peer)
    thread=threading.Thread(target=serve);thread.start()
    results=[]
    try:
        with sync_playwright() as pw:
            for enabled in (False,True,False):
                args=['--disable-background-networking']
                if enabled:args+=['--webrtc-ip-handling-policy=disable_non_proxied_udp','--force-webrtc-ip-handling-policy=disable_non_proxied_udp']
                browser=pw.chromium.launch(channel='chrome',headless=True,args=args)
                try:
                    context=browser.new_context();page=context.new_page()
                    with lock:before=count[0]
                    observed=page.evaluate(SCRIPT,port)
                    with lock:received=count[0]-before
                    item={'network_policy':enabled,'headless':True,'browser':browser.version,'stun_requests':received,**observed}
                    results.append(item)
                    if enabled:assert received==0 and not observed['hasTestMapping'],item
                    else:assert received>0 and observed['hasTestMapping'],item
                finally:browser.close()
            browser=pw.chromium.launch(channel='chrome',headless=True)
            try:
                context=browser.new_context()
                context.add_init_script("Object.defineProperty(window,'RTCPeerConnection',{value:undefined,configurable:false,writable:false});")
                page=context.new_page();page.goto('data:text/html,<script>window.firstObservation=typeof RTCPeerConnection</script>')
                assert page.evaluate('window.firstObservation')=='undefined'
            finally:browser.close()
    finally:
        stop.set();thread.join(timeout=2);sock.close()
    record={'python_environment':platform.system()+' local host; loopback UDP only; no account or Internet STUN','python':platform.python_version(),'playwright':importlib.metadata.version('playwright'),'browser_channel':'chrome','trials':results,'page_api_first_script':'undefined','scope':'One headless Chrome build; artificial TEST-NET mapping; not an anonymity or all-site protection claim'}
    Path(__file__).with_name('verification-results.json').write_text(json.dumps(record,ensure_ascii=False,indent=2)+'\n',encoding='utf-8',newline='\n')
    print(json.dumps(record,ensure_ascii=False,indent=2))


if __name__=='__main__':run()