Step 95. Level 1 종합 프로젝트 — 네트워크 감시 도구 완성하기

Level 1 종합 프로젝트 · 학습 계획 4시간 내외, 개인별 차이 있음

전제: Step 79의 TCP 연결, Step 92~93의 SQLite·매개변수 바인딩, Git 기본 사용. 이번 목표는 요구사항을 읽고 스캔 → 저장 → 리포트 → 비교를 연결하는 것입니다.

같은 조건으로 두 번 관찰했을 때 무엇이 달라졌는지 설명할 수 있는 작은 도구를 만듭니다. 먼저 명세만 보고 구현하고, 막힌 부분은 아래 완전한 참고 구현과 비교하세요. 실습은 본인 컴퓨터의 127.0.0.1에서 진행합니다. 다른 주소를 사용하려면 소유자에게 명시적으로 점검을 허가받은 랩이어야 합니다.

1. 학습 목표

• TCP 연결 성공과 배너 수신을 구분한다.
• 완료한 검사 한 번과 그 결과들을 DB에 함께 저장한다.
• 같은 주소·포트·시간 제한의 직전 검사와 비교한다.
• 첫 검사, 빈 결과, 잘못된 입력, 다른 검사 조건을 재현하고 해석한다.
• 실행 파일·검증 기록·README를 남긴다.

2. 요구사항과 판단 기준

Python 3.10 이상과 표준 라이브러리를 사용합니다. Windows에서는 python, Linux에서는 보통 python3 명령을 사용합니다. 별도 pip 설치는 없습니다. 새 기준 DB는 netwatch-lab.sqlite3이며 기존 예제의 DB를 자동 변환하거나 지우지 않습니다.

해석의 한계: 연결 성공은 검사 시점의 관찰입니다. 연결 실패에는 거절·지연·차단·네트워크 오류가 섞일 수 있습니다. NEW는 이전 같은 조건의 검사에서 관찰되지 않은 연결이며, 서비스가 바로 그 순간 처음 설치됐다는 뜻은 아닙니다. NOT OBSERVED도 영구 폐쇄 판정이 아닙니다.

3. 직접 실행하고 결과 확인하기

3-1. 파일 준비

실습 묶음 ZIP에는 실행 파일, 자동 재현기, 한국어·영어 README, 두 환경의 검증 결과와 파일 해시가 들어 있습니다. 압축을 풀고 그 폴더에서 실행하세요. 개별 파일: netwatch.py · reproduce.py · 한국어 안내.

python netwatch.py --help
python reproduce.py

재현기는 임시 루프백 서버와 시험 DB만 만들고 종료 시 정리합니다. 빈 포트를 자동 선택하므로 포트 번호는 실행마다 달라질 수 있습니다. 이 검사와 별도로 아래 수동 시나리오를 해 보세요.

3-2. 기준선과 변화 없음

터미널 두 개에서 서버를 하나씩 켭니다. 8000이나 9000이 이미 사용 중이면 모든 검사 명령의 포트 목록도 함께 바꾸세요.

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

세 번째 터미널에서 아래 명령을 두 번 실행합니다. 첫 검사부터 8001도 목록에 포함합니다. 첫 결과에는 BASELINE, 두 번째에는 UNCHANGED가 있어야 합니다.

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

3-3. 새 관찰과 사라진 관찰

새 터미널에서 8001 서버를 켜고 같은 검사 명령을 반복하면 [!] NEW: 127.0.0.1:8001이 나옵니다.

python -m http.server 8001 --bind 127.0.0.1

8001 서버를 Ctrl+C로 종료하고 다시 검사하면 [-] NOT OBSERVED: 127.0.0.1:8001을 확인합니다. 실습 뒤 나머지 서버도 종료하세요. 첫 상태를 다시 시험하려면 기존 DB를 지우는 대신 새 --db 파일명을 사용합니다.

3-4. DB에서 독립적으로 확인

Python 3.12 이상은 다음 명령을 지원합니다. 3.10·3.11에서는 Python 대화형에서 sqlite3.connect()와 execute()로 같은 SQL을 실행하세요.

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;"

첫 두 검사에 두 포트, 세 번째에 세 포트, 네 번째에 두 포트가 저장돼야 합니다. HTTP 서버는 요청을 먼저 기다리므로 빈 배너가 정상일 수 있습니다. --json을 붙이면 검사 ID·조건별 이전 ID·오류 집계도 기계가 읽기 쉬운 형태로 확인할 수 있습니다.

4. 미션과 연습문제

참고 구현을 보기 전에 입력 검증, 연결 관찰, 한 검사 단위 저장, 리포트, 조건별 비교를 각각 함수로 나누어 구현하세요. README에는 선택한 범위·대기 제한·종료 방법과 실제 출력의 의미를 적습니다.

1. 처음에는 8000·9000만 요청했다가 다음 검사에 8001을 추가하면 왜 별도의 기준선이 필요한가?
2. 연결은 성공했지만 recv에서 시간이 초과되면 열린 포트 관찰을 버려도 되는가?
3. 세 검사가 저장된 뒤 첫 검사 ID를 조회할 때 최신 두 검사만 비교하면 무엇이 잘못되는가?
4. scans 한 행은 들어갔지만 findings 저장이 실패하면 어떻게 처리해야 하는가?

5. 완전한 참고 구현과 검증 기록

아래 코드는 다운로드 파일과 같습니다. DB 생성과 CLI 진입점까지 포함합니다. 주소 범위를 문자열 끝부분만 바꾸어 만들지 않고 IPv4 주소로 해석한 뒤 같은 /24와 순서를 검사합니다.

"""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())

이번 재현에서 확인한 결과

2026-09-25에 Windows Python 3.12.10와 외부 네트워크·공개 포트가 없는 Linux 컨테이너 Python 3.14.7에서 17개 검증 항목을 각각 통과했습니다. Windows는 임시 포트, Linux는 루프백 8000·8001·9000을 사용했습니다. 아래는 Linux 재현기가 실제 CLI 결과를 저장한 기록의 요약입니다.

추가로 잘못된 CLI 입력 7경우, 배너의 SQL 문자가 데이터로 보존되는지, 연결 후 실제 RST, 빈 기준선, 다른 검사 조건, 지정한 과거 검사, 저장 실패의 전체 취소, 다른 스키마 DB 보존을 확인했습니다. 이는 두 실행 환경의 재현 기록이며 모든 OS·방화벽·네트워크 조건을 검증한 것은 아닙니다. Windows 기록 · Linux 기록.

연습문제 해답

1. 이전에는 8001을 관찰하지 않았으므로 ‘새로 열렸다’고 비교할 근거가 없습니다. 검사 조건이 같아야 변화와 관찰 범위 변경을 구분할 수 있습니다.
2. 버리면 안 됩니다. 요청을 기다리는 서비스는 연결을 수락해도 인사말을 보내지 않을 수 있습니다. 연결 성공과 수신 결과를 별개로 기록합니다.
3. 요청한 첫 검사와 관계없는 시점을 비교하게 됩니다. 해당 scan_id와 동일 조건의 더 이전 완료 검사를 명시적으로 선택해야 합니다.
4. 부분 검사로 남기지 않도록 하나의 트랜잭션에서 헤더와 결과를 함께 커밋하거나 함께 롤백합니다.

6. 흔한 문제와 확인 순서

• 아무 출력도 없음: 함수 조각만 저장했는지 확인하세요. 이 파일 끝에는 main()을 호출하는 진입점이 있습니다.
• lastrowid가 이상함: conn.execute()도 커서를 반환합니다. INSERT를 실행한 바로 그 커서의 lastrowid를 읽으세요. executemany() 뒤에 마지막 개별 행 번호를 기대하지 않습니다.
• 계속 BASELINE: DB 파일·주소 범위·포트 목록·시간 제한이 같은지 확인합니다. 작업 폴더를 바꾸면 상대 DB 경로도 달라집니다.
• 느리거나 미관찰: 주소·포트 수와 대기 시간을 먼저 확인합니다. 거절이 즉시 오는 경우와 무응답으로 시간 제한까지 기다리는 경우는 다릅니다. 짧은 대기는 지연된 연결을 놓칠 수 있습니다.
• 입력 오류: 역순·다른 /24·잘못된 주소는 수정해서 다시 실행합니다. 종료 코드 2와 사용법 안내가 정상적인 거부 결과입니다.
• 기존 DB 스키마 오류: 원래 파일을 보존하고 새 --db 이름을 선택하세요. 이 예제는 자동 마이그레이션을 제공하지 않습니다.

7. 완료 기준과 다음 복습

네 수동 시나리오와 저장된 DB가 일치하고, 비교 조건·빈 배너·실패 관찰의 의미를 본인 말로 설명하면 이번 프로젝트를 마칠 수 있습니다. 실행 결과, 실패한 경우와 수정 이유를 README·커밋에 남기세요. 소켓은 Step79, 저장·바인딩은 Step92~93으로 돌아가 복습합니다.

공식 근거: Python socket · Python sqlite3 · Python ipaddress. 이번 개정은 Step095의 참고 구현과 두 언어 설명을 검토한 것이며 전체 교육과정 검증은 아닙니다.