Step 95. Level 1 Comprehensive Project — Completing a Network Watchdog Tool

Level 1 comprehensive project · Planning guide: about four hours; individual time varies.

Prerequisites: TCP connections from Step79, SQLite and parameter binding from Steps92–93, and basic Git. Connect scan → store → report → compare from a requirements specification.

Build a small tool that explains what changed between observations made under the same conditions. Try the specification first, then compare difficult parts with the complete reference implementation below. This exercise uses 127.0.0.1 on your own computer. Other targets must belong to a lab you own or have explicit permission to assess.

1. Learning objectives

• Separate a successful TCP connection from banner reception.
• Store one completed scan and its findings together.
• Compare with the previous scan of the same addresses, ports and timeout.
• Reproduce a first scan, empty results, invalid input and changed observation conditions.
• Keep runnable files, verification records and a README.

2. Requirements and interpretation

Use Python 3.10+ and the standard library. Commands below use python; on Linux, use python3 if appropriate. No pip packages are required. The new default DB is netwatch-lab.sqlite3. Older example databases are neither migrated nor deleted automatically.

Limits: a connection is a point-in-time observation. A failed connect may mean refusal, delay, filtering or another network error. NEW means a connection was not observed in the previous comparable scan; it does not establish when a service was installed. NOT OBSERVED is not proof of permanent closure.

3. Run it and inspect the results

3-1. Get the files

The lab ZIP contains the runnable program, reproduction script, Korean/English READMEs, results from two environments and file hashes. Extract it and run from that directory. Individual files: netwatch.py · reproduce.py · English instructions.

python netwatch.py --help
python reproduce.py

The reproducer creates temporary loopback servers and a test DB, then closes and removes them. It selects temporary ports, so their numbers can differ between runs. Also try the manual sequence below.

3-2. Baseline and no change

Start one server in each of two terminals. If either port is already occupied, choose alternatives and change the scan’s port list consistently.

python -m http.server 8000 --bind 127.0.0.1
python -m http.server 9000 --bind 127.0.0.1

In a third terminal, run the following twice. Include 8001 in the requested list from the first scan. Expect BASELINE first, then UNCHANGED.

python netwatch.py 127.0.0.1 127.0.0.1 --ports 8000,8001,9000 --db practice.sqlite3

3-3. A new observation and a lost observation

In another terminal, start the 8001 server. Repeat the unchanged scan command and expect [!] NEW: 127.0.0.1:8001.

python -m http.server 8001 --bind 127.0.0.1

Stop that server with Ctrl+C, scan again, and expect [-] NOT OBSERVED: 127.0.0.1:8001. Stop the other servers when finished. To repeat a first-run exercise, select a new --db filename instead of deleting an existing database.

3-4. Check independently in the DB

Python 3.12+ supports these commands. On 3.10–3.11, execute the same SQL through sqlite3.connect() and execute() in the Python interpreter.

python -m sqlite3 practice.sqlite3 "SELECT id,scan_time FROM scans ORDER BY id;"
python -m sqlite3 practice.sqlite3 "SELECT scan_id,ip,port,banner FROM findings ORDER BY scan_id,port;"

Expect two findings in each of the first two scans, three in the third, and two in the fourth. An HTTP server may wait for a request, so an empty banner can be normal. Add --json to inspect the scan ID, previous comparable ID and error counts in machine-readable form.

4. Mission and exercises

Before reading the reference, implement input validation, connection observation, storage of a complete scan, reporting and comparison as separate functions. Document the chosen scope, timeouts, shutdown procedure and interpretation in your README.

1. Why is a new baseline needed if the first requested port list contains only 8000/9000 and the next adds 8001?
2. May a successful connection be discarded because recv times out?
3. What goes wrong if a request for the first scan ID compares only the latest two scans after three scans have been saved?
4. What should happen if the scans row is inserted but storing findings fails?

5. Complete reference and verification records

This code matches the downloadable file and includes database setup and a CLI entry point. It parses IPv4 addresses, checks order and the /24 boundary, and then enumerates the range.

"""Step 095 reference: bounded TCP observations for your own authorized lab.

Python 3.10+; standard library only. No packets are sent on import.
"""
import argparse
from contextlib import closing
from datetime import datetime, timezone
import ipaddress
import json
import math
from pathlib import Path
import socket
import sqlite3
import sys

DEFAULT_PORTS = '22,80,443,445,3389,8000,8001,9000'
SCHEMA = {
    'scans': ['id', 'scan_time', 'scope', 'connection_errors'],
    'findings': ['id', 'scan_id', 'ip', 'port', 'banner'],
}


def make_ip_range(start, end):
    """Validate an ascending IPv4 range inside one /24, then enumerate it."""
    first, last = ipaddress.IPv4Address(start), ipaddress.IPv4Address(end)
    if int(first) > int(last):
        raise ValueError('start IP must not exceed end IP')
    if int(first) >> 8 != int(last) >> 8:
        raise ValueError('both IPs must belong to the same /24')
    return [str(ipaddress.IPv4Address(n)) for n in range(int(first), int(last) + 1)]


def parse_ports(text):
    """Return sorted unique TCP port numbers, with at most 64 entries."""
    parts = text.split(',')
    if not parts or any(not p.isascii() or not p.isdecimal() for p in parts):
        raise ValueError('ports must be comma-separated decimal integers')
    ports = sorted(set(map(int, parts)))
    if len(ports) > 64 or not all(1 <= p <= 65535 for p in ports):
        raise ValueError('use 1 to 64 ports, each between 1 and 65535')
    return ports


def scope_key(ips, ports, timeout):
    """Keep different target ranges, port lists and timeouts out of one baseline."""
    return json.dumps({'start': ips[0], 'end': ips[-1], 'ports': ports,
                       'timeout': timeout, 'probe': 'tcp-connect-v1'}, sort_keys=True)


def scan(ips, ports, timeout):
    """Observe successful TCP connects; a banner failure does not undo a connect."""
    found, errors = [], {}
    for ip in ips:
        for port in ports:
            with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
                sock.settimeout(timeout)
                try:
                    sock.connect((ip, port))
                except OSError as error:
                    key = type(error).__name__
                    errors[key] = errors.get(key, 0) + 1
                    continue
                try:
                    banner = sock.recv(64).decode('utf-8', 'replace').strip()
                except OSError:
                    banner = ''
                found.append((ip, port, banner))
    return found, errors


def initialize(conn):
    """Create a new lesson DB, or refuse an unrelated/older schema unchanged."""
    tables = {r[0] for r in conn.execute(
        "SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%'")}
    if tables:
        if tables != set(SCHEMA):
            raise ValueError('existing DB is not this lesson schema; choose a new --db file')
        for table, columns in SCHEMA.items():
            actual = [r[1] for r in conn.execute('PRAGMA table_info(' + table + ')')]
            if actual != columns:
                raise ValueError('existing DB schema differs; choose a new --db file')
        return
    conn.executescript('''
        CREATE TABLE scans(
            id INTEGER PRIMARY KEY, scan_time TEXT NOT NULL,
            scope TEXT NOT NULL, connection_errors TEXT NOT NULL);
        CREATE TABLE findings(
            id INTEGER PRIMARY KEY, scan_id INTEGER NOT NULL REFERENCES scans(id),
            ip TEXT NOT NULL, port INTEGER NOT NULL, banner TEXT NOT NULL,
            UNIQUE(scan_id, ip, port));
    ''')


def save_scan(conn, scope, found, errors):
    """Commit one complete scan and all findings together, or roll back both."""
    with conn:
        cur = conn.execute(
            'INSERT INTO scans(scan_time,scope,connection_errors) VALUES(?,?,?)',
            (datetime.now(timezone.utc).isoformat(), scope, json.dumps(errors, sort_keys=True)))
        scan_id = cur.lastrowid
        conn.executemany('INSERT INTO findings(scan_id,ip,port,banner) VALUES(?,?,?,?)',
                         [(scan_id, ip, port, banner) for ip, port, banner in found])
    return scan_id


def comparison(conn, scan_id):
    """Compare this specific scan with the previous completed scan of its scope."""
    row = conn.execute('SELECT scope FROM scans WHERE id=?', (scan_id,)).fetchone()
    if row is None:
        raise ValueError('scan ID does not exist')
    previous = conn.execute(
        'SELECT id FROM scans WHERE scope=? AND id<? ORDER BY id DESC LIMIT 1',
        (row[0], scan_id)).fetchone()
    def endpoints(sid):
        return set(conn.execute('SELECT ip,port FROM findings WHERE scan_id=?', (sid,)))
    now = endpoints(scan_id)
    if previous is None:
        return {'previousScanId': None, 'baseline': True, 'new': [], 'lost': []}
    before = endpoints(previous[0])
    return {'previousScanId': previous[0], 'baseline': False,
            'new': sorted(now - before), 'lost': sorted(before - now)}


def report(conn, scan_id, errors):
    """Read this scan's observations from the DB and attach a bounded comparison."""
    devices = {}
    for ip, port in conn.execute(
            'SELECT ip,port FROM findings WHERE scan_id=? ORDER BY ip,port', (scan_id,)):
        devices.setdefault(ip, []).append(port)
    return {'scanId': scan_id, 'devices': devices,
            'observedOpenCount': sum(map(len, devices.values())),
            'connectionErrors': errors, **comparison(conn, scan_id)}


def show(result):
    print(f"Responding IP addresses: {len(result['devices'])}; observed open TCP ports: {result['observedOpenCount']}")
    for ip, ports in result['devices'].items():
        print('  ' + ip + ': ' + ', '.join(map(str, ports)))
    if result['baseline']:
        print('BASELINE: first completed scan for this target/port/timeout scope')
    elif not result['new'] and not result['lost']:
        print('UNCHANGED: same observed endpoints as the previous comparable scan')
    else:
        for ip, port in result['new']:
            print(f'[!] NEW: {ip}:{port}')
        for ip, port in result['lost']:
            print(f'[-] NOT OBSERVED: {ip}:{port}')
    if result['connectionErrors']:
        print('Connect failures: ' + json.dumps(result['connectionErrors'], sort_keys=True))


def main(argv=None):
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('start'); parser.add_argument('end')
    parser.add_argument('--ports', default=DEFAULT_PORTS)
    parser.add_argument('--timeout', type=float, default=0.3)
    parser.add_argument('--db', type=Path, default=Path('netwatch-lab.sqlite3'))
    parser.add_argument('--json', action='store_true', help='print a machine-readable report')
    args = parser.parse_args(argv)
    try:
        ips = make_ip_range(args.start, args.end)
        ports = parse_ports(args.ports)
        if not math.isfinite(args.timeout) or not 0.05 <= args.timeout <= 2:
            raise ValueError('timeout must be a finite number between 0.05 and 2 seconds')
    except ValueError as error:
        parser.error(str(error))
    try:
        with closing(sqlite3.connect(args.db)) as conn:
            conn.execute('PRAGMA foreign_keys=ON')
            initialize(conn)
            found, errors = scan(ips, ports, args.timeout)
            scan_id = save_scan(conn, scope_key(ips, ports, args.timeout), found, errors)
            result = report(conn, scan_id, errors)
        print(json.dumps(result, ensure_ascii=True)) if args.json else show(result)
        return 0
    except (OSError, sqlite3.Error, ValueError) as error:
        print('ERROR: ' + str(error), file=sys.stderr)
        return 1
    except KeyboardInterrupt:
        print('Interrupted; check the DB for the last committed scan.', file=sys.stderr)
        return 130


if __name__ == '__main__':
    sys.exit(main())

Observed reproduction results

On 2026-09-25, all 17 verification items passed on Windows with Python 3.12.10 and in a Linux container with Python 3.14.7, no external network and no published ports. Windows used temporary ports; Linux used loopback 8000/8001/9000. The table summarizes actual CLI results saved by the Linux reproducer.

Additional checks covered seven invalid CLI inputs, preserving SQL characters in a banner as data, a real reset after connect, an empty baseline, changed observation conditions, a specified historical scan, transaction rollback on storage failure and preserving an unrelated DB. These are records from two environments, not verification of every OS, firewall or network. Windows results · Linux results.

Exercise answers

1. The earlier scan did not observe 8001 at all. Identical observation conditions are needed to separate a state change from a change in coverage.
2. No. A service may accept TCP while waiting silently for a request. Keep the connection result separate from the receive result.
3. It compares the wrong points in time. Select the requested scan_id and an earlier completed scan with matching conditions.
4. Commit the header and findings in one transaction, or roll both back. Do not leave a partial scan as a completed baseline.

6. Troubleshooting

• No output: check that you saved the complete program rather than function fragments. The last lines call main().
• Unexpected lastrowid: conn.execute() returns a cursor too. Read lastrowid from the cursor that performed the INSERT; do not expect executemany() to provide the last individual row ID.
• Repeated BASELINE: check the DB path, range, port list and timeout. Changing directories also changes the meaning of a relative DB path.
• Slow or missing observations: check address/port count and timeout. Immediate refusal differs from waiting for no reply. Short timeouts can miss delayed connections.
• Input error: correct reversed ranges, different /24s or malformed addresses. Exit code 2 and usage text are the intended refusal.
• Different DB schema: keep the old file and choose a new –db filename. This example does not perform automatic migration.

7. Completion and review

Finish when the four manual scenarios agree with the saved DB and you can explain comparison conditions, empty banners and failed observations. Keep actual outputs, failed cases and your corrections in the README and commits. Review sockets in Step79 and storage/binding in Steps92–93.

Official references: Python socket · Python sqlite3 · Python ipaddress. This revision reviews the Step095 reference and its two language explanations; it does not certify the entire curriculum.