NPM를 통한 접속 로그 실시간으로 알림 받기

 외부에서 나만 접속하기 위한 웹이 있을 것이다. 예를 들어 DSM, file browser, webdav라던지

wireguard나 tailscale 같은 vpn을 통해 외부에 공개하지 않고 내부망으로 접속하는 것이 보안적으로 가장 좋은 방식이지만 아무래도 특정기기만 가능하고 다른 기기에서 일회성으로 접속하기엔 힘들어 외부에 공개하지만 나만 쓰고 싶은 웹이 있을 것이다.

나의 경우에는 filebrowser가 그에 해당한다.

 

알다시피 Nginx는 리버스 프록시 서버의 역할을 한다. 내 filebrowser에 접속하려면 도메인을 통해 Nginx를 무조건 거치게 된다.

IP를 통해 접속하려고 해도 접속이 불가능하다.

공인IP:443으로 접속했을때 화면

 

당연하게도 Nginx, NPM에는 접속 로그가 남게 된다.

그러면 파이썬 스크립트를 통해 접속 로그를 실시간으로 읽어 나에게 알림을 보내준다면 보안적으로 더욱 좋을 것이다.

내가 접속하지 않았는데 접근 알림이 온다면 다른 사람이 접근했다는 이야기가 되고 그 사람이 어떠한 공격을 할지 모르니 서버를 끄거나 외부 포트를 막아버리는 차단을 할 수 있으니 말이다.

 

알림은 디스코드 웹훅으로 받을 예정이다.

 

여기에 GeoIP를 추가하여 알림을 받는다면 어디 나라 또는 asn에서 접속하는지 알림을 받을 수 있다.

먼저 리눅스 시스템에 파이썬이 설치되어 있고 GeoIP를 다운로드하기 위한 MaxMind에 가입되어 있으며 디스코드 계정이 있다는 가정으로 시작한다.

 

필수 라이브러리를 먼저 설치한다.

pip install requests geoip2

 

GeoIP 목록은 주기적으로 업데이트를 해줘야 한다.

목록을 다운 및 업데이트를 하기 위해 대부분의 리눅스 배포판에 패키지가 존재한다.

sudo apt install geoipupdate

 

GeoIP.conf를 수정하여 라이선스 키를 입력해야 한다.

sudo nano /etc/GeoIP.conf

 

# MaxMind 계정 ID (숫자로 된 ID)
AccountID [여기에_MaxMind_계정_ID_입력]

# 1단계에서 발급받은 라이선스 키
LicenseKey [여기에_라이선스_키_입력]

# 다운로드할 데이터베이스 버전
EditionIDs [ "GeoLite2-Country", "GeoLite2-ASN" ]

# (중요!) 데이터베이스를 저장할 경로
DatabaseDirectory /opt/npm/geoip

 

데이터베이스 절대경로는 변경은 가능하지만 변경 시 아래의 스크립트의 경로도 모두 바꿔야 하기 때문에 유지하는 것을 추천한다.

 

아래 명령어 입력 시 목록이 다운 및 업데이트된다.

sudo geoipupdate -v

 

명령어 실행 후 /opt/npm/geoip 디렉터리에는 GeoLite2-Country.mmdb와 GeoLite2-ASN.mmdb가 다운로드하여져 있을 것이다.

 

(권장)

매번 목록을 업데이트하기 위해 관리자가 geoipupdate -v 명령어를 입력하기는 불편하고 어렵다.

그래서 cron을 이용해 자동화를 하려고 한다.

업데이트 명령의 위치는 /usr/bin/geoipupdate이다.

업데이트 주기는 1주일에 한 번을 권장한다.

너무 잦은 주기로 업데이트를 하면 차단을 당하기 때문이다.

 

crontab 명령어로 자동화를 해준다.

crontab -e

0 4 * * 0 /usr/bin/geoipupdate

 

이렇게 되면 매주 일요일 4시에 데이터베이스가 업데이트된다.

 

이제 원하는 경로에 디렉터리를 만든 후 원하는 이름의. py를 만든다.

스크립트의 코드는 아래와 같다.

import sys
import re
import requests
import geoip2.database

WEBHOOK_URL = ""  # 여기에 디스코드 웹훅 URL을 붙여넣으세요.
COUNTRY_DB_PATH = '/opt/npm/geoip/GeoLite2-Country.mmdb'
ASN_DB_PATH = '/opt/npm/geoip/GeoLite2-ASN.mmdb'

def get_geoip_info(ip_address):
    country_code = "N/A"
    asn_name = "N/A"
    try:
        with geoip2.database.Reader(COUNTRY_DB_PATH) as reader:
            response = reader.country(ip_address)
            country_code = response.country.iso_code
    except Exception:
        pass
    try:
        with geoip2.database.Reader(ASN_DB_PATH) as reader:
            response = reader.asn(ip_address)
            asn_name = response.autonomous_system_organization
    except Exception:
        pass
    return country_code, asn_name

def send_to_discord(log_line):
    ip_match = re.search(r'\[Client\s+([^\]]+)\]', log_line)
    status_match = re.search(r'\]\s+-\s+(\d{3})', log_line)
    request_match = re.search(r'-\s+((?:GET|POST|PUT|DELETE|PATCH|HEAD|OPTIONS)\s+https?.*?)"', log_line)

    if not ip_match:
        return

    ip_address = ip_match.group(1)
    status_code = status_match.group(1) if status_match else "N/A"
    request_info = request_match.group(1).strip() if request_match else "N/A"

    country, asn = get_geoip_info(ip_address)

    embed = {
        "title": f"🌐 New Access Log ({status_code})",
        "color": 5814783 if status_code.startswith('2') else 15158332,
        "fields": [
            {"name": "Country", "value": f":flag_{country.lower() if country != 'N/A' else 'white'}: {country}", "inline": True},
            {"name": "ASN", "value": asn, "inline": False},
            {"name": "Request", "value": f"`{request_info}`", "inline": False},
        ]
    }
    payload = {"embeds": [embed]}
    try:
        requests.post(WEBHOOK_URL, json=payload)
    except Exception as e:
        print(f"웹훅 전송 실패: {e}")

if __name__ == "__main__":
    for line in sys.stdin:
        send_to_discord(line.strip())

 소스코드의 정규식은 NPM 기준이기 때문에 Nginx의 설정에 따라 달라질 수 있다.

 

소스코드에 디스코드 웹훅을 입력해야 하는데 웹훅 생성 방법은 아래와 같다.

디스코드 실행 후 개인 서버를 생성한다.

개인서버 생성 후 채팅 채널을 개설 한 뒤 채널 편집에 들어간다.

연동 탭에 웹후크 탭이 있을 것이다. 새 웹후크를 누른 후 이름을 설정 한 뒤 웹후크 URL 복사 버튼을 눌러 복사한 후 스크립트에 붙여놓는다.

 

작성 후 스크립트를 실행하려면 내가 확인하려는 로그의 절대 경로를 알아야 한다.

NPM 기준 로그는 /data/logs 디렉터리에 있다.

내 경우는 프록시 서버를 거치기 때문에 proxy-host-1_access.log를 확인하면 된다.

 

리눅스에는 tail -F 명령어를 통해 특정 파일을 계속해서 추적할 수 있다. -n 0 옵션을 통해 새 로그 라인만 확인하게 만든다.

파이프라인을 통해 새 로그를 파이썬 스크립트로 넘기면 된다.

tail -F -n 0 /data/logs/proxy-host-1_access.log | python3 <파이썬 스크립트 절대경로>
#확인 해야하는 로그가 2개 이상이라면 추가로 절대 경로를 입력하면 된다.
#tail -F -n 0 /data/logs/proxy-host-1_access.log /data/logs/proxy-host-2_access.log | python3 /path/to/your_script.py

 

새 로그가 들어올 때마다 tail명령어가 실행되면서 뒤에 스크립트가 실행될 것이다.

 

그럼 여기서 문제점이 하나 생긴다.

바로 재부팅을 하면 tail 명령어를 다시 입력해야 한다는 것이다.

NPM 관리자는 이거 조차도 귀찮을 것이다.

로그 확인이 힘들어 스크립트를 만들었는데 스크립트 실행은 직접 해야 한다니 말이 안 되는 것이다.

 

그래서 systemd에 파이썬 스크립트를 등록하여 enable 명령어로 재부팅 시 자동 시작을 하게 만들면 된다.

 

/etc/systemd/system 아래 이름.service로 등록하면 된다.

 

sudo nano /etc/systemd/system/name.service

 

내용은 아래와 같다.

[Unit]
Description=NPM Access Log to Discord Webhook
# 네트워크가 준비된 후에 실행되도록 설정
After=network.target

[Service]
Type=simple
#파이썬 스크립트의 절대경로가 올바르게 입력되야함
ExecStart=/bin/bash -c "/usr/bin/tail -F -n 0 /data/logs/proxy-host-1_access.log /data/logs/proxy-host-2_access.log | /usr/bin/python3 /path/to/your_script.py"

# 서비스가 비정상 종료되면 항상 다시 시작
Restart=always

[Install]
# 다중 사용자 모드(일반 부팅)에서 이 서비스를 활성화
WantedBy=multi-user.target

 

파일 저장이 완료되면 아래의 명령어를 순서대로 입력한다.

# 1. systemd에 새로 만든 서비스 파일 인식시키기
sudo systemctl daemon-reload

# 2. 재부팅 시 이 서비스가 자동으로 시작되도록 활성화
sudo systemctl enable name.service

# 3. 지금 당장 서비스를 시작
sudo systemctl start name.service

 

리눅스 사용자라면 자주 보는 명령어 일 것이다.

status를 통해 잘 실행 중인지 확인한다.

systemctl status name.service

 

이렇게 모든 설정이 완료 됐다면 도메인을 통해 웹 접속을 해보자

디스코드 웹훅 알림

 

사진과 같이 알림이 잘 오는 것을 확인할 수 있다.

하지만 모든 알림이 계속 오기 때문에 무조건 이전에 업로드한 opnsense 외국 접근 차단과 추후에 업로드할 Cloudflare WAF를 통한 국가 및 봇 차단을 설정하고 스크립트를 설정하는 것을 추천한다. 만약 다른 설정을 하지 않으면 봇의 접속과 외국 접속 때문에 알림 굉장히 많이 오게 된다.