psutil을 이용한 특정 프로세스 네트워크 세션(ESTABLISHED) 수집

시스템 운영 및 모니터링 도구를 개발할 때, 특정 사용자가 실행한 프로세스가 어떤 외부 IP와 통신 중인지 확인해야 하는 경우가 있다. 이 글에서는 psutil을 사용하여 특정 프로세스를 필터링하고, 현재 TCP 연결이 수립(ESTABLISHED)된 세션 정보만 추출하는 방법을 알아본다.

 

1. psutil 개요

psutil (Python System and Process Utilities)은 실행 중인 프로세스와 시스템 활용률(CPU, 메모리, 디스크, 네트워크 등) 정보를 수집하는 크로스 플랫폼 라이브러리다. 시스템 모니터링, 프로파일링, 프로세스 관리 도구 개발에 주로 사용된다.

• 설치: pip install psutil

 

2. 코드 예시

다음은 사용자(user), 프로세스명(process_name), 감시 대상 IP 목록(target_ips)을 입력받아, 조건에 맞는 활성 연결 정보를 반환하는 함수다.

import psutil
from typing import List

def get_psutil_connections(self, user: str, process_name: str, target_ips: List[str]) -> List[dict]:
    connections = []
    target_ip_set = set(target_ips)

    for proc in psutil.process_iter(['pid', 'username', 'name']):
        try:
            # 사용자 및 프로세스 이름 필터링
            if proc.info['username'] != user or process_name not in proc.info['name']:
                continue

            # TCP 연결 정보 조회
            for conn in proc.connections(kind='tcp'):
                # 4. 연결 상태가 ESTABLISHED인 경우만
                if conn.status != "ESTABLISHED":
                    continue

                # IPv4-mapped IPv6 주소(::ffff:) 처리 및 문자열 변환
                laddr = f"{conn.laddr.ip}:{conn.laddr.port}".replace('::ffff:', '') if conn.laddr else ""
                raddr = f"{conn.raddr.ip}:{conn.raddr.port}".replace('::ffff:', '') if conn.raddr else ""

                # 타겟 IP 포함 여부 확인
                if laddr in target_ip_set or raddr in target_ip_set:
                    connections.append({
                        "laddr": laddr, "raddr": raddr
                    })
        except (psutil.NoSuchProcess, psutil.AccessDenied):
            # 프로세스 종료 혹은 권한 부족 시 무시
            continue

    return connections

 

3. 주요 메서드 및 파라미터 상세

위 코드에서 사용된 psutil의 메서드와 속성은 다음과 같다.

1) psutil.process_iter(attrs=None, ad_value=None)

실행 중인 모든 프로세스를 순회하는 Iterator를 반환한다.

• attrs (List[str]):
  • ['pid', 'username', 'name']과 같이 필요한 속성 리스트를 인자로 전달하면, proc.info 딕셔너리에 해당 정보를 담아 반환한다.

{'username': 'root', 'name': 'systemd', 'pid': 1}
{'username': 'root', 'name': 'kthreadd', 'pid': 2}
...

 

2) proc.connections(kind='inet')

해당 프로세스(proc)와 연관된 소켓 연결 정보를 리스트로 반환한다.

• kind (str): 조회할 프로토콜 유형을 지정한다.
  • "tcp": TCP 연결만 조회 (본 예제에서 사용)
  • "udp": UDP 연결만 조회
  • "inet": IPv4 및 IPv6
  • "all": UNIX 소켓을 포함한 모든 연결
• 반환 객체: pconn이라는 NamedTuple을 반환하며 주요 속성은 다음과 같다.
  • laddr: 로컬 주소 (ip, port)
  • raddr: 원격 주소 (ip, port)
  • status: 연결 상태 문자열

 

3) conn.status

TCP 연결의 상태를 나타내는 문자열이다.

• "ESTABLISHED": 연결이 수립되어 데이터 전송이 가능한 상태.
• "LISTEN": 포트가 열려있고 연결 요청을 대기하는 상태.
• "TIME_WAIT", "CLOSE_WAIT": 연결 종료 과정의 상태.
• 위 코드는 실질적인 통신이 이루어지는 세션을 찾기 위해 "ESTABLISHED" 상태만 필터링.

 

4) conn.laddr 및 conn.raddr

• laddr (Local Address): 로컬 머신의 IP와 포트.
• raddr (Remote Address): 연결된 원격지의 IP와 포트.
• 데이터 처리: Python의 소켓 라이브러리는 종종 IPv4 주소를 IPv6 형식(::ffff:192.168.0.1)으로 표현한다. 정확한 IP 매칭을 위해 replace('::ffff:', '')를 사용하여 순수 IPv4 주소만 추출한다.

 

4. 예외 처리 로직

1. psutil.NoSuchProcess: process_iter가 루프를 도는 도중 프로세스가 종료되면, 상세 정보를 조회할 때 해당 프로세스를 찾을 수 없어 발생한다.
2. psutil.AccessDenied: 스크립트를 실행한 사용자가 해당 프로세스(예: root 소유 프로세스)의 정보를 읽을 권한이 없을 때 발생한다.

위 두 가지 예외 상황에서 프로그램이 중단되지 않고 다음 프로세스로 넘어가도록 try-except 구성.