"""Real loopback sockets + independent DB assertions. No external targets. Run in the extracted lesson directory: python reproduce.py Use --fixed-ports only in an isolated environment where 8000/8001/9000 are free. """ from contextlib import closing import hashlib import json import os import platform from pathlib import Path import socket import sqlite3 import struct import subprocess import sys import tempfile import threading import time import netwatch ROOT = Path(__file__).resolve().parent class Fixture: def __init__(self, mode='silent', port=0, active=True): self.mode = mode self.sock = socket.socket() self.sock.bind(('127.0.0.1', port)) self.port = self.sock.getsockname()[1] self.stop = threading.Event() self.thread = None self.handlers = [] if active: self.start() def start(self): self.sock.listen(16) self.sock.settimeout(0.1) self.thread = threading.Thread(target=self.accept, daemon=True) self.thread.start() def accept(self): while not self.stop.is_set(): try: connection, _ = self.sock.accept() except socket.timeout: continue except OSError: break handler = threading.Thread(target=self.handle, args=(connection,), daemon=True) self.handlers.append(handler) handler.start() def handle(self, connection): with connection: try: if self.mode == 'greeting': connection.sendall(b"hello'); DROP TABLE scans; --\r\n") elif self.mode == 'reset': fmt = 'HH' if sys.platform == 'win32' else 'ii' connection.setsockopt(socket.SOL_SOCKET, socket.SO_LINGER, struct.pack(fmt, 1, 0)) else: connection.settimeout(1) connection.recv(1) except OSError: pass def close(self): self.stop.set() self.sock.close() if self.thread: self.thread.join(2) for handler in self.handlers: handler.join(2) def run(): fixed = '--fixed-ports' in sys.argv fixtures = [] checks, scenarios = [], [] def fixture(*args, **kwargs): item = Fixture(*args, **kwargs) fixtures.append(item) return item try: with tempfile.TemporaryDirectory(prefix='netwatch-test-', dir=os.environ.get('NETWATCH_TEST_DIR',ROOT)) as temp: work = Path(temp) db = work/'observations.sqlite3' a = fixture('silent', port=8000 if fixed else 0) b = fixture('greeting', port=9000 if fixed else 0) c = fixture('silent', port=8001 if fixed else 0, active=False) ports = ','.join(map(str, [a.port, b.port, c.port])) args = ['127.0.0.1', '127.0.0.1', '--ports', ports, '--timeout', '0.05', '--db', str(db)] def cli(extra=None, replace=None): command = [sys.executable, str(ROOT/'netwatch.py'), *(replace if replace is not None else args), *(extra or [])] return subprocess.run(command, cwd=work, capture_output=True, text=True, encoding='utf-8', timeout=20) def scenario(label): result = cli(['--json']) assert result.returncode == 0, result.stderr data = json.loads(result.stdout) scenarios.append({'name': label, 'result': data}) return data first = scenario('baseline') assert first['baseline'] and first['observedOpenCount'] == 2 and not first['new'] second = scenario('unchanged') assert not second['baseline'] and not second['new'] and not second['lost'] c.start() third = scenario('new_server') assert third['new'] == [['127.0.0.1', c.port]] and third['observedOpenCount'] == 3 c.close() fourth = scenario('server_stopped') assert fourth['lost'] == [['127.0.0.1', c.port]] and not fourth['new'] checks += ['real_cli_baseline', 'real_cli_unchanged', 'real_cli_new_endpoint', 'real_cli_lost_observation'] with closing(sqlite3.connect(db)) as conn: assert conn.execute('SELECT COUNT(*) FROM scans').fetchone()[0] == 4 assert conn.execute('SELECT COUNT(*) FROM findings').fetchone()[0] == 9 banner = conn.execute('SELECT banner FROM findings WHERE port=?', (b.port,)).fetchone()[0] assert banner == "hello'); DROP TABLE scans; --" assert conn.execute('PRAGMA integrity_check').fetchone()[0] == 'ok' assert netwatch.comparison(conn, first['scanId'])['baseline'] assert netwatch.comparison(conn, third['scanId'])['new'] == [('127.0.0.1', c.port)] count = conn.execute('SELECT COUNT(*) FROM scans').fetchone()[0] try: netwatch.save_scan(conn, 'duplicate-proof', [('127.0.0.1', 7, '')]*2, {}) raise AssertionError('expected unique-constraint failure') except sqlite3.IntegrityError: pass assert conn.execute('SELECT COUNT(*) FROM scans').fetchone()[0] == count assert not conn.execute("SELECT 1 FROM scans WHERE scope='duplicate-proof'").fetchone() checks += ['independent_persistent_rows', 'sql_binding_preserves_literal_banner', 'specific_old_scan_not_latest_two', 'atomic_failed_insert'] changed_scope = args.copy(); changed_scope[changed_scope.index('--ports')+1] = str(a.port) result = cli(['--json'], replace=changed_scope) assert result.returncode == 0 and json.loads(result.stdout)['baseline'] changed_timeout = args.copy(); changed_timeout[changed_timeout.index('--timeout')+1] = '0.06' assert json.loads(cli(['--json'], replace=changed_timeout).stdout)['baseline'] assert not scenario('original_scope_again')['new'] checks += ['different_ports_get_own_baseline', 'different_timeout_gets_own_baseline', 'original_scope_reuses_its_own_previous_scan'] reset = fixture('reset') found, _ = netwatch.scan(['127.0.0.1'], [reset.port], 0.1) assert len(found) == 1 and found[0][1] == reset.port checks += ['real_reset_after_connect_remains_observed_open'] for bad in [[], ['abc', '127.0.0.1'], ['127.0.0.1','127.0.1.2'], ['127.0.0.3','127.0.0.1'], ['127.00.0.1','127.0.0.1'], ['127.0.0.1','127.0.0.1','--ports','0'], ['127.0.0.1','127.0.0.1','--timeout','nan']]: result = cli(replace=bad) assert result.returncode == 2 and 'usage:' in result.stderr.lower() assert not (work/'netwatch-lab.sqlite3').exists() assert netwatch.make_ip_range('127.0.0.1','127.0.0.3') == ['127.0.0.1','127.0.0.2','127.0.0.3'] checks += ['seven_invalid_cli_cases_no_database', 'inclusive_range_validation'] unrelated = work/'unrelated.sqlite3' with closing(sqlite3.connect(unrelated)) as conn: conn.execute('CREATE TABLE personal(value TEXT)');conn.execute('INSERT INTO personal VALUES(?)', ('keep',));conn.commit() before = hashlib.sha256(unrelated.read_bytes()).hexdigest() otherargs = args.copy(); otherargs[otherargs.index('--db')+1] = str(unrelated) result = cli(replace=otherargs) assert result.returncode == 1 and 'choose a new --db file' in result.stderr assert hashlib.sha256(unrelated.read_bytes()).hexdigest() == before checks += ['unrelated_database_refused_without_changes'] empty = fixture(active=False) emptyargs = args.copy();emptyargs[emptyargs.index('--ports')+1] = str(empty.port) result = cli(['--json'], replace=emptyargs) assert result.returncode == 0 and json.loads(result.stdout)['observedOpenCount'] == 0 and json.loads(result.stdout)['baseline'] checks += ['empty_observation_is_a_valid_baseline'] human = cli(); assert human.returncode == 0 and 'UNCHANGED' in human.stdout checks += ['human_report_complete'] return {'status':'pass','at':time.time(),'python':sys.version.split()[0], 'platform':platform.platform(), 'targets':['127.0.0.1'], 'fixedPorts':fixed, 'checks':checks, 'scenarios':scenarios, 'humanOutput':human.stdout, 'productionWrites':False, 'sourceSHA256':{name:hashlib.sha256((ROOT/name).read_bytes()).hexdigest() for name in ['netwatch.py','reproduce.py']}} finally: for item in reversed(fixtures): item.close() if __name__ == '__main__': print(json.dumps(run(), ensure_ascii=True, indent=2))