자체 메일서버 관리 API 코드 공개 (mailapi.py)
·41 분
이 글은 제가 VMware + 우분투 위에 직접 구축한 자체 메일서버를 웹 대시보드에서
관리하기 위해 만든 백엔드 API 서버(mailapi.py) 코드를 공개하는 글입니다.
프레임워크 없이 파이썬 표준 라이브러리(http.server)만으로 만든 아주 작은
API 서버이고, 코딩을 잘 몰라도 흐름을 따라올 수 있도록 함수마다 한글 주석을
달았습니다. 다만 실제 운영 서버의 보안과 직결되는 값들(접근 차단 경로 목록 등)은
***로 가려두었으니, 그대로 복붙해서 쓰기보다는 참고용으로 봐주세요.
전체 코드는 아래와 같습니다.
from http.server import BaseHTTPRequestHandler, HTTPServer, ThreadingHTTPServer
import json
import re
import hmac
import MySQLdb
import subprocess
import os
import shutil
import logging
from logging.handlers import RotatingFileHandler
from datetime import datetime, timezone, timedelta
from urllib.parse import urlparse, parse_qs
import tempfile
import glob
# =====================================================================
# 이 코드는 무엇인가? (초보자를 위한 개요)
# =====================================================================
# 자체 메일서버(postfix + dovecot + MariaDB)를 웹 화면(대시보드)에서
# 관리할 수 있게 해주는 "뒷단(백엔드) API 서버" 코드입니다.
#
# - 프레임워크(Django, Flask 같은 것) 없이, 파이썬 표준 라이브러리에
# 들어있는 http.server만으로 아주 작은 웹서버를 직접 만들었습니다.
# - 사람이 브라우저로 직접 보는 화면(프론트엔드/UI)은 이 파일에 없고,
# 별도의 웹페이지(PHP)가 이 서버에 "데이터 좀 줘/바꿔줘"라고
# 요청을 보내면, 이 코드가 그 요청을 처리해서 JSON 형식으로 답을
# 돌려주는 구조입니다. 이런 걸 흔히 "REST API 서버"라고 부릅니다.
# - 모든 요청은 헤더에 담긴 비밀 API 키(X-API-Key)가 맞아야만 처리됩니다.
# (_require_api_key 함수가 문지기 역할)
# - do_GET / do_POST 두 함수가 "이 URL로 요청 오면 이 함수 실행해라"
# 라고 나눠주는 교통정리 역할을 합니다. (예: GET /account/list →
# handle_account_list 함수 실행)
# - 계정/도메인/별칭 같은 데이터는 MariaDB(MySQL 호환 DB)에 저장하고,
# 실제 메일 파일 자체는 리눅스 폴더(/var/mail/vhosts/도메인/계정)에
# 저장합니다. 이 둘을 함께 맞춰주는 코드가 곳곳에 있습니다.
# - 보안/운영 점검용 기능(로그 조회, 보안 점검, 파일탐색기, 백업)도
# 포함되어 있어서, 서버에 SSH로 직접 들어가지 않고도 웹 화면에서
# 웬만한 관리 작업을 할 수 있게 만든 것이 이 프로젝트의 목적입니다.
#
# ※ 이 글은 학습/공유 목적으로 일부 보안에 민감한 값(***)을 가려서
# 공개합니다. 실제 운영 시에는 이 코드를 그대로 복붙하지 말고,
# 본인 환경에 맞게 검토 후 사용하세요.
# =====================================================================
DB_CONFIG = {
"host": os.environ.get("MAILAPI_DB_HOST", "127.0.0.1"),
"user": os.environ.get("MAILAPI_DB_USER", "mailadmin"),
"passwd": os.environ.get("MAILAPI_DB_PASS", ""),
"db": os.environ.get("MAILAPI_DB_NAME", "mailserver"),
"charset": "utf8mb4",
}
MAIL_BASE_DIR = os.environ.get("MAILAPI_MAIL_BASE_DIR", "/var/mail/vhosts")
MAIL_UID = int(os.environ.get("MAILAPI_MAIL_UID", "1005"))
MAIL_GID = int(os.environ.get("MAILAPI_MAIL_GID", "1005"))
DOVECOT_SCHEME = os.environ.get("MAILAPI_DOVECOT_SCHEME", "SHA512-CRYPT")
API_BIND_HOST = os.environ.get("MAILAPI_BIND_HOST", "127.0.0.1")
API_BIND_PORT = int(os.environ.get("MAILAPI_BIND_PORT", "18080"))
API_KEY = os.environ.get("MAILAPI_API_KEY", "").strip()
AUDIT_LOG_PATH = os.environ.get("MAILAPI_AUDIT_LOG", "/var/log/mailadmin-api.log")
AUDIT_LOG_MAX_BYTES = int(os.environ.get("MAILAPI_AUDIT_MAX_BYTES", str(5 * 1024 * 1024)))
AUDIT_LOG_BACKUP_COUNT = int(os.environ.get("MAILAPI_AUDIT_BACKUP_COUNT", "5"))
DKIM_BASE_DIR = os.environ.get("MAILAPI_DKIM_BASE_DIR", "/etc/opendkim/keys")
DKIM_DEFAULT_SELECTOR = os.environ.get("MAILAPI_DKIM_DEFAULT_SELECTOR", "default")
DKIM_KEY_BITS = int(os.environ.get("MAILAPI_DKIM_KEY_BITS", "2048"))
BACKUP_BASE_DIR = os.environ.get("MAILAPI_BACKUP_DIR", "/var/backups/mailadmin")
MAILAPI_SQL_DUMP_BIN = os.environ.get("MAILAPI_SQL_DUMP_BIN", "mysqldump")
MAILAPI_SQL_BIN = os.environ.get("MAILAPI_SQL_BIN", "mysql")
SYSTEMD_SERVICES = [
s.strip()
for s in os.environ.get("MAILAPI_SYSTEMD_SERVICES", "postfix,dovecot,mariadb").split(",")
if s.strip()
]
# 파일탐색기: 여기 정의된 루트 밖은 절대 접근 불가 (경로 조작으로도 못 벗어남)
CERT_LIVE_DIR = os.environ.get("MAILAPI_CERT_LIVE_DIR", "/etc/letsencrypt/live")
# 원클릭으로 볼 수 있는 서버 로그 목록 — 경로가 다르면 환경변수로 개별 override 가능
LOG_SOURCES = {
"mail": {"type": "file", "path": os.environ.get("MAILAPI_LOG_MAIL", "/var/log/mail.log"), "label": "메일 로그 (Postfix/Dovecot)"},
"mail_err": {"type": "file", "path": os.environ.get("MAILAPI_LOG_MAIL_ERR", "/var/log/mail.err"), "label": "메일 에러 로그"},
"fail2ban": {"type": "file", "path": os.environ.get("MAILAPI_LOG_FAIL2BAN", "/var/log/fail2ban.log"), "label": "fail2ban 로그"},
"auth": {"type": "file", "path": os.environ.get("MAILAPI_LOG_AUTH", "/var/log/auth.log"), "label": "인증 로그 (SSH 등)"},
"syslog": {"type": "file", "path": os.environ.get("MAILAPI_LOG_SYSLOG", "/var/log/syslog"), "label": "시스템 로그"},
"mariadb": {"type": "journal", "unit": os.environ.get("MAILAPI_LOG_MARIADB_UNIT", "mariadb"), "label": "MariaDB 로그 (journalctl)"},
"letsencrypt": {"type": "file", "path": os.environ.get("MAILAPI_LOG_LETSENCRYPT", "/var/log/letsencrypt/letsencrypt.log"), "label": "Let's Encrypt 갱신 로그"},
"mailadmin_api": {"type": "file", "path": AUDIT_LOG_PATH, "label": "Mail Admin API 감사로그"},
"php_fpm": {"type": "file", "path": os.environ.get("MAILAPI_LOG_PHP_FPM", "/var/log/php8.3-fpm.log"), "label": "PHP-FPM 마스터 로그"},
"php_errors": {"type": "file", "path": os.environ.get("MAILAPI_LOG_PHP_ERRORS", "/var/log/php_errors.log"), "label": "PHP 스크립트 에러 로그"},
"php_ui_journal": {"type": "file", "path": os.environ.get("MAILAPI_LOG_UI_ACCESS", "/var/log/nginx/mailadmin-access.log"), "label": "Mail Admin UI 접속 로그 (nginx)"},
"mailadmin_error": {"type": "file", "path": os.environ.get("MAILAPI_LOG_UI_ERROR", "/var/log/nginx/mailadmin-error.log"), "label": "Mail Admin UI 에러 로그 (nginx)"},
"kern": {"type": "file", "path": os.environ.get("MAILAPI_LOG_KERN", "/var/log/kern.log"), "label": "커널 로그"},
"dpkg": {"type": "file", "path": os.environ.get("MAILAPI_LOG_DPKG", "/var/log/dpkg.log"), "label": "패키지 설치/업데이트 로그"},
"cron_jobs": {"type": "file", "path": os.environ.get("MAILAPI_LOG_CRON_JOBS", "/var/log/cron_jobs.log"), "label": "크론잡 실행 로그"},
"tempmail_abuse": {"type": "file", "path": os.environ.get("MAILAPI_LOG_TEMPMAIL_ABUSE", "/var/log/tempmail/abuse.log"), "label": "임시메일 어뷰징 로그"},
"tempmail_expire": {"type": "file", "path": os.environ.get("MAILAPI_LOG_TEMPMAIL_EXPIRE", "/var/log/tempmail/expire.log"), "label": "임시메일 만료 처리 로그"},
"tempmail_access": {"type": "file", "path": os.environ.get("MAILAPI_LOG_TEMPMAIL_ACCESS", "/var/log/nginx/tempmail-access.log"), "label": "임시메일 nginx 접속 로그"},
"tempmail_error": {"type": "file", "path": os.environ.get("MAILAPI_LOG_TEMPMAIL_ERROR", "/var/log/nginx/tempmail-error.log"), "label": "임시메일 nginx 에러 로그"},
"mailadmin_ui_auth": {"type": "file", "path": os.environ.get("MAILAPI_LOG_UI_AUTH", "/var/log/mailadmin-ui-auth.log"), "label": "Mail Admin UI 로그인 실패 로그"},
}
LOG_TAIL_MAX_BYTES = 2 * 1024 * 1024 # 파일 끝에서 최대 2MB만 읽어서 tail 계산 (대용량 로그 보호)
FILE_MANAGER_ROOT_PATH = "/"
# [보안 마스킹]
# 파일탐색기 기능이 "접근 금지"로 취급하는 실제 경로 목록입니다.
# 이 값을 그대로 공개하면 "여기만 피하면 뚫린다"는 지도를 넘겨주는 셈이라
# 블로그 공개용 코드에서는 실제 값 대신 자리표시자(***)로 가려두었습니다.
# 실제로 운영할 때는 시스템 계정 정보(/etc/shadow, /etc/passwd 등),
# SSH 설정, 크론, systemd, PAM/보안 설정, /root, /home 등
# "건드리면 서버 전체가 위험해지는 경로"를 최대한 넉넉하게 등록해두세요.
FILE_MANAGER_BLOCKED_PATHS = [
p.strip()
for p in os.environ.get(
"MAILAPI_FILEMANAGER_BLOCKED",
"***,***,***,***,***" # 예: 계정정보 / SSH / 크론 / systemd / 보안설정 등 (실제 값은 비공개)
).split(",")
if p.strip()
]
FILE_MANAGER_QUICKLINKS = [
{"path": "/", "label": "전체 (/)"},
{"path": "/etc/postfix", "label": "Postfix 설정"},
{"path": "/etc/dovecot", "label": "Dovecot 설정"},
{"path": "/etc/opendkim", "label": "OpenDKIM 설정"},
{"path": "/opt/mailadmin", "label": "mailadmin (API/UI)"},
{"path": "/var/log", "label": "로그"},
]
FILE_MANAGER_MAX_VIEW_BYTES = 15 * 1024 * 1024 # 15MB 넘는 파일은 보기 자체를 거부
FILE_MANAGER_MAX_EDIT_BYTES = 3 * 1024 * 1024 # 3MB 넘으면 보기는 되지만 저장은 거부
def setup_audit_logger():
# 모든 API 요청/결과를 파일에 기록하는 '감사로그' 준비 함수. 나중에 누가 언제 뭘 했는지 되짚어볼 때 씀.
logger = logging.getLogger("mailadmin_audit")
logger.setLevel(logging.INFO)
if logger.handlers:
return logger
handler = RotatingFileHandler(
AUDIT_LOG_PATH,
maxBytes=AUDIT_LOG_MAX_BYTES,
backupCount=AUDIT_LOG_BACKUP_COUNT,
encoding="utf-8",
)
formatter = logging.Formatter("%(message)s")
handler.setFormatter(formatter)
logger.addHandler(handler)
return logger
AUDIT_LOGGER = setup_audit_logger()
class MailAdminHandler(BaseHTTPRequestHandler):
server_version = "MailAdminAPI/2.0"
def log_message(self, format, *args):
# 파이썬 기본 웹서버가 콘솔에 남기는 접속 로그를 그대로 쓰겠다는 뜻(커스터마이즈 안 함).
super().log_message(format, *args)
def _send_json(self, code, data):
# 결과를 JSON 형식으로 만들어서 브라우저/앱에 응답으로 돌려주는 공용 함수. 모든 API가 이걸 씀.
body = json.dumps(data, ensure_ascii=False).encode("utf-8")
self.send_response(code)
self.send_header("Content-Type", "application/json; charset=utf-8")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
def _read_json(self):
# 요청 본문(body)에 담겨온 JSON 데이터를 파이썬 딕셔너리로 읽어들이는 함수.
length = int(self.headers.get("Content-Length", "0"))
if length <= 0:
return {}
raw = self.rfile.read(length)
return json.loads(raw.decode("utf-8"))
def _get_db(self):
# MariaDB(=MySQL)에 접속하는 함수. DB_CONFIG에 적힌 계정정보로 연결한다.
return MySQLdb.connect(**DB_CONFIG)
def _hash_password(self, plain_password):
# 사용자가 입력한 평문 비밀번호를 메일서버(dovecot)가 이해하는 암호화된 형태로 바꿔주는 함수.
result = subprocess.run(
["doveadm", "pw", "-s", DOVECOT_SCHEME, "-p", plain_password],
capture_output=True,
text=True,
check=True,
)
return result.stdout.strip()
def _normalize_active(self, value):
# true/1/'1' 등 여러 형태로 들어오는 '활성화 여부' 값을 0 또는 1로 통일해주는 함수.
if value in (True, "true", "True", "1", 1):
return 1
return 0
def _split_email(self, email):
# 'user@example.com' 같은 이메일을 로컬파트(user)와 도메인(example.com)으로 쪼개는 함수.
email = email.strip().lower()
if "@" not in email:
return None, None
localpart, domain = email.rsplit("@", 1)
if not localpart or not domain:
return None, None
return localpart, domain
def _domain_path(self, domain):
# 한 도메인의 메일이 저장될 폴더 경로를 계산하는 함수 (예: /var/mail/vhosts/example.com).
return os.path.join(MAIL_BASE_DIR, domain)
def _maildir_path(self, domain, localpart):
# 특정 계정의 메일이 저장될 폴더 경로를 계산하는 함수.
return os.path.join(MAIL_BASE_DIR, domain, localpart)
def _ensure_domain_dir(self, domain):
# 도메인용 메일 폴더가 없으면 새로 만들고, 권한(소유자/모드)까지 맞춰주는 함수.
domain_dir = self._domain_path(domain)
os.makedirs(domain_dir, exist_ok=True)
os.chown(domain_dir, MAIL_UID, MAIL_GID)
os.chmod(domain_dir, 0o755)
return domain_dir
def _ensure_maildir(self, domain, localpart):
# 계정용 메일함 폴더(cur/new/tmp 하위폴더 포함)를 만들고 권한을 세팅하는 함수. 메일서버 표준 규격(Maildir)을 따른다.
mailbox_dir = self._maildir_path(domain, localpart)
os.makedirs(mailbox_dir, exist_ok=True)
os.chown(mailbox_dir, MAIL_UID, MAIL_GID)
os.chmod(mailbox_dir, 0o700)
for subdir in ("cur", "new", "tmp"):
path = os.path.join(mailbox_dir, subdir)
os.makedirs(path, exist_ok=True)
os.chown(path, MAIL_UID, MAIL_GID)
os.chmod(path, 0o700)
return mailbox_dir
def _remove_maildir(self, domain, localpart):
# 계정 삭제 시 그 계정의 메일 폴더를 통째로 지우는 함수.
mailbox_dir = self._maildir_path(domain, localpart)
if os.path.isdir(mailbox_dir):
shutil.rmtree(mailbox_dir)
return True
return False
def _client_ip(self):
# 이 요청을 보낸 클라이언트(호출한 쪽)의 IP 주소를 가져오는 함수.
return self.client_address[0] if self.client_address else "-"
def _audit(self, action, target="", status_code=200, result="ok", detail=""):
# '누가, 언제, 뭘, 성공/실패했는지'를 한 줄 기록으로 남기는 함수. 모든 API 처리 끝에 호출된다.
record = {
"ts": datetime.now(timezone.utc).isoformat(),
"client_ip": self._client_ip(),
"method": self.command,
"path": self.path,
"action": action,
"target": target,
"status_code": status_code,
"result": result,
"detail": detail,
}
AUDIT_LOGGER.info(json.dumps(record, ensure_ascii=False))
def _require_api_key(self):
# 요청 헤더에 담긴 API 키가 서버가 알고 있는 키와 일치하는지 검사하는 '문지기' 함수. 여기서 막히면 이후 로직은 실행되지 않는다.
if not API_KEY:
self._audit("auth", "-", 500, "error", "api_key_not_configured")
self._send_json(
500,
{
"status": "error",
"message": "api key is not configured",
"error": "api_key_not_configured",
}
)
return False
client_key = self.headers.get("X-API-Key", "").strip()
if not client_key:
self._audit("auth", "-", 401, "error", "missing_api_key")
self._send_json(
401,
{
"status": "error",
"message": "missing api key",
"error": "missing_api_key",
}
)
return False
if not hmac.compare_digest(client_key, API_KEY):
self._audit("auth", "-", 403, "error", "invalid_api_key")
self._send_json(
403,
{
"status": "error",
"message": "invalid api key",
"error": "invalid_api_key",
}
)
return False
return True
def _run(self, cmd, env=None, check=True):
# 외부 명령어(리눅스 명령)를 실행하고 결과를 돌려받는 공용 함수.
return subprocess.run(
cmd,
capture_output=True,
text=True,
check=check,
env=env,
)
def _maildir_size_bytes(self, path):
# 메일함 폴더 안 모든 파일의 용량을 더해서 총 사용량(바이트)을 계산하는 함수.
total = 0
if not os.path.isdir(path):
return 0
for root, dirs, files in os.walk(path):
for name in files:
try:
fp = os.path.join(root, name)
if not os.path.islink(fp):
total += os.path.getsize(fp)
except OSError:
pass
return total
def _human_size(self, n):
# 1234567 같은 바이트 숫자를 '1.18 MB'처럼 사람이 읽기 쉬운 형태로 바꿔주는 함수.
value = float(n)
units = ["B", "KB", "MB", "GB", "TB"]
for unit in units:
if value < 1024.0 or unit == units[-1]:
return f"{value:.2f} {unit}"
value /= 1024.0
return f"{n} B"
def _dkim_domain_dir(self, domain):
# DKIM(메일 위조 방지용 서명) 키가 저장된 도메인별 폴더 경로를 계산하는 함수.
return os.path.join(DKIM_BASE_DIR, domain)
def _dkim_private_key_path(self, domain, selector):
# DKIM 개인키 파일 경로를 계산하는 함수.
return os.path.join(self._dkim_domain_dir(domain), f"{selector}.private")
def _dkim_txt_path(self, domain, selector):
# DNS에 등록해야 할 DKIM 공개키(TXT 레코드) 파일 경로를 계산하는 함수.
return os.path.join(self._dkim_domain_dir(domain), f"{selector}.txt")
def _read_file_text(self, path):
# 텍스트 파일을 열어서 내용을 통째로 읽어오는 공용 함수.
with open(path, "r", encoding="utf-8", errors="replace") as f:
return f.read()
def _mysql_env(self):
# mysql/mysqldump 명령을 실행할 때 비밀번호를 환경변수로 넘기기 위한 준비 함수.
env = os.environ.copy()
env["MYSQL_PWD"] = DB_CONFIG["passwd"]
return env
def _fm_resolve(self, rel_path):
# 파일탐색기에서 받은 경로가 '접근 금지 목록'에 걸리는지 검사하고, 안전하면 실제 경로를 돌려주는 함수. 심볼릭 링크로 우회하는 것도 막는다.
"""경로를 안전하게 검증한다. 블랙리스트에 걸리면 ValueError.
심볼릭 링크로 블랙리스트를 우회하는 것도 os.path.realpath로 실제 경로를 계산해 막는다."""
rel_path = (rel_path or "").strip()
candidate = os.path.realpath(os.path.join(FILE_MANAGER_ROOT_PATH, rel_path.lstrip("/")))
for blocked in FILE_MANAGER_BLOCKED_PATHS:
blocked_real = os.path.realpath(blocked)
if candidate == blocked_real or candidate.startswith(blocked_real + os.sep):
raise ValueError("path_blocked")
return candidate
def do_GET(self):
# GET 방식 요청(주로 '조회'용)이 들어왔을 때, URL 경로를 보고 어떤 처리 함수로 넘길지 나눠주는 '교통정리' 함수.
if not self._require_api_key():
return
parsed = urlparse(self.path)
path_only = parsed.path
if path_only == "/health":
self._audit("health", "-", 200, "ok", "health_check")
self._send_json(
200,
{
"status": "ok",
"message": "mailadmin api is running",
"bind": f"{API_BIND_HOST}:{API_BIND_PORT}",
"auth": "x-api-key",
}
)
return
if path_only == "/domain/list":
return self.handle_domain_list()
if path_only == "/account/list":
return self.handle_account_list()
if path_only == "/account/search":
return self.handle_account_search(parsed)
if path_only == "/account/quota":
return self.handle_account_quota(parsed)
if path_only == "/audit/log":
return self.handle_audit_log(parsed)
if path_only == "/dkim/list":
return self.handle_dkim_list()
if path_only == "/dkim/public":
return self.handle_dkim_public(parsed)
if path_only == "/backup/list":
return self.handle_backup_list()
if path_only == "/alias/list":
return self.handle_alias_list()
if path_only == "/system/status":
return self.handle_system_status()
if path_only == "/system/fail2ban":
return self.handle_fail2ban_status()
if path_only == "/system/certs":
return self.handle_cert_status()
if path_only == "/tempmail/stats":
return self.handle_tempmail_stats()
if path_only == "/security/checks":
return self.handle_security_checks(parsed)
if path_only == "/logs/sources":
return self.handle_log_sources()
if path_only == "/logs/tail":
return self.handle_log_tail(parsed)
if path_only == "/file/roots":
return self.handle_file_roots()
if path_only == "/file/list":
return self.handle_file_list(parsed)
if path_only == "/file/read":
return self.handle_file_read(parsed)
self._audit("unknown", "-", 404, "error", "not_found")
self._send_json(
404,
{
"status": "error",
"message": "endpoint not found",
"error": "not_found",
"path": self.path,
}
)
def do_POST(self):
# POST 방식 요청(주로 '생성/변경'용)이 들어왔을 때, URL 경로를 보고 어떤 처리 함수로 넘길지 나눠주는 '교통정리' 함수.
if not self._require_api_key():
return
try:
data = self._read_json()
except json.JSONDecodeError:
self._audit("json_decode", "-", 400, "error", "invalid_json")
self._send_json(
400,
{
"status": "error",
"message": "invalid json",
"error": "invalid_json",
}
)
return
if self.path == "/domain/add":
return self.handle_domain_add(data)
if self.path == "/domain/delete":
return self.handle_domain_delete(data)
if self.path == "/domain/set-tempmail":
return self.handle_domain_set_tempmail(data)
if self.path == "/account/add":
return self.handle_account_add(data)
if self.path == "/account/delete":
return self.handle_account_delete(data)
if self.path == "/account/password":
return self.change_password(data)
if self.path == "/account/set-active":
return self.handle_account_set_active(data)
if self.path == "/account/test-login":
return self.handle_account_test_login(data)
if self.path == "/dkim/generate":
return self.handle_dkim_generate(data)
if self.path == "/backup/create":
return self.handle_backup_create(data)
if self.path == "/backup/restore":
return self.handle_backup_restore(data)
if self.path == "/alias/add":
return self.handle_alias_add(data)
if self.path == "/alias/delete":
return self.handle_alias_delete(data)
if self.path == "/alias/set-active":
return self.handle_alias_set_active(data)
if self.path == "/system/fail2ban/unban":
return self.handle_fail2ban_unban(data)
if self.path == "/file/write":
return self.handle_file_write(data)
self._audit("unknown", "-", 404, "error", "not_found")
self._send_json(
404,
{
"status": "error",
"message": "endpoint not found",
"error": "not_found",
"path": self.path,
}
)
def handle_domain_add(self, data):
# 새 메일 도메인을 DB에 등록하고, 그 도메인용 폴더도 함께 만드는 API.
conn = None
cur = None
domain = data.get("domain", "").strip().lower()
try:
if not domain:
self._audit("domain_add", "-", 400, "error", "domain_required")
self._send_json(400, {"status": "error", "message": "domain is required", "error": "domain_required"})
return
conn = self._get_db()
cur = conn.cursor()
cur.execute("SELECT id FROM virtual_domains WHERE name=%s", (domain,))
row = cur.fetchone()
if row:
self._audit("domain_add", domain, 409, "error", "domain_exists")
self._send_json(409, {"status": "error", "message": "domain already exists", "error": "domain_exists", "domain": domain})
return
cur.execute("INSERT INTO virtual_domains (name) VALUES (%s)", (domain,))
conn.commit()
domain_dir = self._ensure_domain_dir(domain)
self._audit("domain_add", domain, 200, "ok", domain_dir)
self._send_json(200, {"status": "ok", "message": "domain added", "domain": domain, "mail_root": domain_dir})
except Exception as e:
self._audit("domain_add", domain or "-", 500, "error", str(e))
self._send_json(500, {"status": "error", "message": "server error", "error": "server_error", "detail": str(e)})
finally:
if cur:
cur.close()
if conn:
conn.close()
def handle_domain_delete(self, data):
# 도메인을 삭제하는 API. 단, 그 도메인에 계정이 하나라도 남아있으면 거부한다(먼저 계정부터 지우게 유도).
conn = None
cur = None
domain = data.get("domain", "").strip().lower()
try:
if not domain:
self._audit("domain_delete", "-", 400, "error", "domain_required")
self._send_json(400, {"status": "error", "message": "domain is required", "error": "domain_required"})
return
conn = self._get_db()
cur = conn.cursor()
cur.execute("SELECT id FROM virtual_domains WHERE name=%s", (domain,))
row = cur.fetchone()
if not row:
self._audit("domain_delete", domain, 404, "error", "domain_not_found")
self._send_json(404, {"status": "error", "message": "domain not found", "error": "domain_not_found", "domain": domain})
return
domain_id = row[0]
cur.execute("SELECT COUNT(*) FROM virtual_users WHERE domain_id=%s", (domain_id,))
user_count = cur.fetchone()[0]
if user_count > 0:
self._audit("domain_delete", domain, 409, "error", f"domain_has_users:{user_count}")
self._send_json(
409,
{
"status": "error",
"message": "domain has users",
"error": "domain_has_users",
"domain": domain,
"user_count": user_count,
}
)
return
cur.execute("DELETE FROM virtual_domains WHERE id=%s", (domain_id,))
conn.commit()
self._audit("domain_delete", domain, 200, "ok", "domain_deleted")
self._send_json(200, {"status": "ok", "message": "domain deleted", "domain": domain})
except Exception as e:
self._audit("domain_delete", domain or "-", 500, "error", str(e))
self._send_json(500, {"status": "error", "message": "server error", "error": "server_error", "detail": str(e)})
finally:
if cur:
cur.close()
if conn:
conn.close()
def handle_domain_set_tempmail(self, data):
# 특정 도메인을 '임시메일용으로 쓸지 말지' 켜고 끄는 API.
conn = None
cur = None
domain = data.get("domain", "").strip().lower()
try:
enabled = self._normalize_active(data.get("enabled", 0))
if not domain:
self._audit("domain_set_tempmail", "-", 400, "error", "domain_required")
self._send_json(400, {"status": "error", "message": "domain is required", "error": "domain_required"})
return
conn = self._get_db()
cur = conn.cursor()
cur.execute("UPDATE virtual_domains SET tempmail_enabled=%s WHERE name=%s", (enabled, domain))
if cur.rowcount == 0:
self._audit("domain_set_tempmail", domain, 404, "error", "domain_not_found")
self._send_json(404, {"status": "error", "message": "domain not found", "error": "domain_not_found", "domain": domain})
return
conn.commit()
self._audit("domain_set_tempmail", domain, 200, "ok", f"tempmail_enabled={enabled}")
self._send_json(200, {"status": "ok", "message": "tempmail flag updated", "domain": domain, "tempmail_enabled": enabled})
except Exception as e:
self._audit("domain_set_tempmail", domain or "-", 500, "error", str(e))
self._send_json(500, {"status": "error", "message": "server error", "error": "server_error", "detail": str(e)})
finally:
if cur:
cur.close()
if conn:
conn.close()
def handle_domain_list(self):
# 등록된 모든 도메인과, 각 도메인에 몇 개 계정이 있는지 목록으로 보여주는 API.
conn = None
cur = None
try:
conn = self._get_db()
cur = conn.cursor()
cur.execute(
"""
SELECT
d.id,
d.name,
COUNT(u.id) AS user_count,
d.tempmail_enabled
FROM virtual_domains d
LEFT JOIN virtual_users u ON u.domain_id = d.id
GROUP BY d.id, d.name, d.tempmail_enabled
ORDER BY d.name ASC
"""
)
rows = cur.fetchall()
items = []
for row in rows:
items.append({
"id": row[0],
"domain": row[1],
"user_count": int(row[2]),
"tempmail_enabled": int(row[3]) if row[3] is not None else 0,
})
self._audit("domain_list", "-", 200, "ok", f"count={len(items)}")
self._send_json(200, {"status": "ok", "count": len(items), "domains": items})
except Exception as e:
self._audit("domain_list", "-", 500, "error", str(e))
self._send_json(500, {"status": "error", "message": "server error", "error": "server_error", "detail": str(e)})
finally:
if cur:
cur.close()
if conn:
conn.close()
def handle_account_add(self, data):
# 새 메일 계정(이메일 주소)을 만드는 API. DB에 등록 + 비밀번호 암호화 + 메일함 폴더 생성까지 한 번에 처리한다.
conn = None
cur = None
email = data.get("email", "").strip().lower()
try:
password = data.get("password", "")
active = self._normalize_active(data.get("active", 1))
if not email:
self._audit("account_add", "-", 400, "error", "email_required")
self._send_json(400, {"status": "error", "message": "email is required", "error": "email_required"})
return
localpart, domain = self._split_email(email)
if not localpart or not domain:
self._audit("account_add", email, 400, "error", "invalid_email")
self._send_json(400, {"status": "error", "message": "invalid email format", "error": "invalid_email", "email": email})
return
if not password:
self._audit("account_add", email, 400, "error", "password_required")
self._send_json(400, {"status": "error", "message": "password is required", "error": "password_required"})
return
conn = self._get_db()
cur = conn.cursor()
cur.execute("SELECT id FROM virtual_domains WHERE name=%s", (domain,))
domain_row = cur.fetchone()
if not domain_row:
self._audit("account_add", email, 404, "error", "domain_not_found")
self._send_json(404, {"status": "error", "message": "domain not found", "error": "domain_not_found", "domain": domain})
return
domain_id = domain_row[0]
cur.execute("SELECT id FROM virtual_users WHERE email=%s", (email,))
if cur.fetchone():
self._audit("account_add", email, 409, "error", "account_exists")
self._send_json(409, {"status": "error", "message": "account already exists", "error": "account_exists", "email": email})
return
password_hash = self._hash_password(password)
cur.execute(
"INSERT INTO virtual_users (domain_id, password, email, active) VALUES (%s, %s, %s, %s)",
(domain_id, password_hash, email, active)
)
conn.commit()
self._ensure_domain_dir(domain)
mailbox_dir = self._ensure_maildir(domain, localpart)
self._audit("account_add", email, 200, "ok", f"active={active},maildir={mailbox_dir}")
self._send_json(
200,
{
"status": "ok",
"message": "account added",
"email": email,
"domain": domain,
"active": active,
"maildir": mailbox_dir,
}
)
except subprocess.CalledProcessError as e:
detail = e.stderr.strip() if e.stderr else str(e)
self._audit("account_add", email or "-", 500, "error", f"password_hash_failed:{detail}")
self._send_json(500, {"status": "error", "message": "password hash failed", "error": "password_hash_failed", "detail": detail})
except Exception as e:
self._audit("account_add", email or "-", 500, "error", str(e))
self._send_json(500, {"status": "error", "message": "server error", "error": "server_error", "detail": str(e)})
finally:
if cur:
cur.close()
if conn:
conn.close()
def handle_account_delete(self, data):
# 메일 계정을 삭제하는 API. DB 기록과 실제 메일함 폴더를 함께 지운다.
conn = None
cur = None
email = data.get("email", "").strip().lower()
try:
if not email:
self._audit("account_delete", "-", 400, "error", "email_required")
self._send_json(400, {"status": "error", "message": "email is required", "error": "email_required"})
return
localpart, domain = self._split_email(email)
if not localpart or not domain:
self._audit("account_delete", email, 400, "error", "invalid_email")
self._send_json(400, {"status": "error", "message": "invalid email format", "error": "invalid_email", "email": email})
return
conn = self._get_db()
cur = conn.cursor()
cur.execute("SELECT id FROM virtual_users WHERE email=%s", (email,))
row = cur.fetchone()
db_deleted = False
if row:
cur.execute("DELETE FROM virtual_users WHERE email=%s", (email,))
conn.commit()
db_deleted = True
maildir_removed = self._remove_maildir(domain, localpart)
if not db_deleted and not maildir_removed:
self._audit("account_delete", email, 404, "error", "account_not_found")
self._send_json(404, {"status": "error", "message": "account not found", "error": "account_not_found", "email": email})
return
self._audit("account_delete", email, 200, "ok", f"db_deleted={db_deleted},maildir_removed={maildir_removed}")
self._send_json(
200,
{
"status": "ok",
"message": "account deleted",
"email": email,
"db_deleted": db_deleted,
"maildir_removed": maildir_removed,
}
)
except Exception as e:
self._audit("account_delete", email or "-", 500, "error", str(e))
self._send_json(500, {"status": "error", "message": "server error", "error": "server_error", "detail": str(e)})
finally:
if cur:
cur.close()
if conn:
conn.close()
def handle_account_list(self):
# 등록된 모든 메일 계정 목록을 보여주는 API.
conn = None
cur = None
try:
conn = self._get_db()
cur = conn.cursor()
cur.execute(
"""
SELECT
u.id,
u.email,
d.name AS domain,
u.active
FROM virtual_users u
JOIN virtual_domains d ON u.domain_id = d.id
ORDER BY u.email ASC
"""
)
rows = cur.fetchall()
items = []
for row in rows:
items.append({"id": row[0], "email": row[1], "domain": row[2], "active": int(row[3])})
self._audit("account_list", "-", 200, "ok", f"count={len(items)}")
self._send_json(200, {"status": "ok", "count": len(items), "accounts": items})
except Exception as e:
self._audit("account_list", "-", 500, "error", str(e))
self._send_json(500, {"status": "error", "message": "server error", "error": "server_error", "detail": str(e)})
finally:
if cur:
cur.close()
if conn:
conn.close()
def handle_account_search(self, parsed):
# 이메일 주소나 도메인, 활성화 여부로 계정을 검색하는 API.
conn = None
cur = None
try:
qs = parse_qs(parsed.query)
q = qs.get("q", [""])[0].strip().lower()
domain = qs.get("domain", [""])[0].strip().lower()
active_raw = qs.get("active", [""])[0].strip()
where = []
params = []
if q:
where.append("u.email LIKE %s")
params.append(f"%{q}%")
if domain:
where.append("d.name = %s")
params.append(domain)
if active_raw in ("0", "1"):
where.append("u.active = %s")
params.append(int(active_raw))
sql = """
SELECT
u.id,
u.email,
d.name AS domain,
u.active
FROM virtual_users u
JOIN virtual_domains d ON u.domain_id = d.id
"""
if where:
sql += " WHERE " + " AND ".join(where)
sql += " ORDER BY u.email ASC"
conn = self._get_db()
cur = conn.cursor()
cur.execute(sql, tuple(params))
rows = cur.fetchall()
items = []
for row in rows:
items.append({"id": row[0], "email": row[1], "domain": row[2], "active": int(row[3])})
self._audit("account_search", q or domain or "-", 200, "ok", f"count={len(items)}")
self._send_json(
200,
{
"status": "ok",
"count": len(items),
"accounts": items,
"filters": {
"q": q,
"domain": domain,
"active": active_raw,
}
}
)
except Exception as e:
self._audit("account_search", "-", 500, "error", str(e))
self._send_json(500, {"status": "error", "message": "server error", "error": "server_error", "detail": str(e)})
finally:
if cur:
cur.close()
if conn:
conn.close()
def handle_account_set_active(self, data):
# 계정을 켜고(활성화) 끄는(비활성화) API. 바꾼 뒤에는 로그인 캐시도 함께 비워준다.
conn = None
cur = None
email = data.get("email", "").strip().lower()
try:
active = self._normalize_active(data.get("active", 0))
if not email:
self._audit("account_set_active", "-", 400, "error", "email_required")
self._send_json(400, {"status": "error", "message": "email is required", "error": "email_required"})
return
localpart, domain = self._split_email(email)
if not localpart or not domain:
self._audit("account_set_active", email, 400, "error", "invalid_email")
self._send_json(400, {"status": "error", "message": "invalid email format", "error": "invalid_email", "email": email})
return
conn = self._get_db()
cur = conn.cursor()
cur.execute("UPDATE virtual_users SET active=%s WHERE email=%s", (active, email))
if cur.rowcount == 0:
self._audit("account_set_active", email, 404, "error", "account_not_found")
self._send_json(404, {"status": "error", "message": "account not found", "error": "account_not_found", "email": email})
return
conn.commit()
cache_flushed = False
cache_flush_error = None
try:
self._run(["doveadm", "auth", "cache", "flush", email])
cache_flushed = True
except subprocess.CalledProcessError as e:
cache_flush_error = e.stderr.strip() if e.stderr else str(e)
except Exception as e:
cache_flush_error = str(e)
self._audit(
"account_set_active",
email,
200,
"ok",
f"active={active},cache_flushed={cache_flushed}" + (f",cache_flush_error={cache_flush_error}" if cache_flush_error else "")
)
self._send_json(
200,
{
"status": "ok",
"message": "account active updated",
"email": email,
"active": active,
"auth_cache_flushed": cache_flushed,
}
)
except Exception as e:
self._audit("account_set_active", email or "-", 500, "error", str(e))
self._send_json(500, {"status": "error", "message": "server error", "error": "server_error", "detail": str(e)})
finally:
if cur:
cur.close()
if conn:
conn.close()
def change_password(self, data):
# 계정 비밀번호를 바꾸는 API. 바꾼 뒤에는 로그인 캐시도 함께 비워준다.
conn = None
cur = None
email = data.get("email", "").strip().lower()
try:
password = data.get("password", "")
if not email or not password:
self._audit("account_password", email or "-", 400, "error", "missing_parameters")
self._send_json(400, {"status": "error", "message": "email and password are required", "error": "missing_parameters"})
return
localpart, domain = self._split_email(email)
if not localpart or not domain:
self._audit("account_password", email, 400, "error", "invalid_email")
self._send_json(400, {"status": "error", "message": "invalid email format", "error": "invalid_email", "email": email})
return
pw_hash = self._hash_password(password)
conn = self._get_db()
cur = conn.cursor()
cur.execute("UPDATE virtual_users SET password=%s WHERE email=%s", (pw_hash, email))
if cur.rowcount == 0:
self._audit("account_password", email, 404, "error", "account_not_found")
self._send_json(404, {"status": "error", "message": "account not found", "error": "account_not_found", "email": email})
return
conn.commit()
cache_flushed = False
cache_flush_error = None
try:
self._run(["doveadm", "auth", "cache", "flush", email])
cache_flushed = True
except subprocess.CalledProcessError as e:
cache_flush_error = e.stderr.strip() if e.stderr else str(e)
except Exception as e:
cache_flush_error = str(e)
self._audit(
"account_password",
email,
200,
"ok",
"password_changed" + (f",cache_flush_error={cache_flush_error}" if cache_flush_error else ",cache_flushed=True")
)
self._send_json(200, {"status": "ok", "message": "password changed", "email": email, "auth_cache_flushed": cache_flushed})
except subprocess.CalledProcessError as e:
detail = e.stderr.strip() if e.stderr else str(e)
self._audit("account_password", email or "-", 500, "error", f"password_hash_failed:{detail}")
self._send_json(500, {"status": "error", "message": "password hash failed", "error": "password_hash_failed", "detail": detail})
except Exception as e:
self._audit("account_password", email or "-", 500, "error", str(e))
self._send_json(500, {"status": "error", "message": "server error", "error": "server_error", "detail": str(e)})
finally:
if cur:
cur.close()
if conn:
conn.close()
def handle_account_test_login(self, data):
# 이메일/비밀번호로 실제 로그인이 되는지 서버에서 직접 테스트해보는 API (문제 진단용).
email = data.get("email", "").strip().lower()
password = data.get("password", "")
try:
if not email or not password:
self._audit("account_test_login", email or "-", 400, "error", "missing_parameters")
self._send_json(400, {"status": "error", "message": "email and password are required", "error": "missing_parameters"})
return
localpart, domain = self._split_email(email)
if not localpart or not domain:
self._audit("account_test_login", email, 400, "error", "invalid_email")
self._send_json(400, {"status": "error", "message": "invalid email format", "error": "invalid_email", "email": email})
return
result = self._run(["doveadm", "auth", "test", email, password], check=False)
success = (result.returncode == 0)
detail = (result.stdout or "").strip()
stderr = (result.stderr or "").strip()
self._audit(
"account_test_login",
email,
200,
"ok" if success else "error",
detail if detail else stderr,
)
self._send_json(
200,
{
"status": "ok",
"email": email,
"auth_success": success,
"stdout": detail,
"stderr": stderr,
"returncode": result.returncode,
}
)
except Exception as e:
self._audit("account_test_login", email or "-", 500, "error", str(e))
self._send_json(500, {"status": "error", "message": "server error", "error": "server_error", "detail": str(e)})
def handle_account_quota(self, parsed):
# 각 계정이 메일함을 얼마나 쓰고 있는지(용량) 보여주는 API.
conn = None
cur = None
try:
qs = parse_qs(parsed.query)
domain_filter = qs.get("domain", [""])[0].strip().lower()
email_filter = qs.get("email", [""])[0].strip().lower()
sql = """
SELECT
u.id,
u.email,
d.name AS domain,
u.active
FROM virtual_users u
JOIN virtual_domains d ON u.domain_id = d.id
"""
where = []
params = []
if domain_filter:
where.append("d.name = %s")
params.append(domain_filter)
if email_filter:
where.append("u.email = %s")
params.append(email_filter)
if where:
sql += " WHERE " + " AND ".join(where)
sql += " ORDER BY u.email ASC"
conn = self._get_db()
cur = conn.cursor()
cur.execute(sql, tuple(params))
rows = cur.fetchall()
items = []
for row in rows:
email = row[1]
localpart, domain = self._split_email(email)
maildir = self._maildir_path(domain, localpart)
size_bytes = self._maildir_size_bytes(maildir)
items.append(
{
"id": row[0],
"email": email,
"domain": row[2],
"active": int(row[3]),
"maildir": maildir,
"size_bytes": size_bytes,
"size_human": self._human_size(size_bytes),
}
)
self._audit("account_quota", domain_filter or email_filter or "-", 200, "ok", f"count={len(items)}")
self._send_json(
200,
{
"status": "ok",
"count": len(items),
"accounts": items,
"filters": {"domain": domain_filter, "email": email_filter},
}
)
except Exception as e:
self._audit("account_quota", "-", 500, "error", str(e))
self._send_json(500, {"status": "error", "message": "server error", "error": "server_error", "detail": str(e)})
finally:
if cur:
cur.close()
if conn:
conn.close()
def handle_audit_log(self, parsed):
# 지금까지 쌓인 '감사로그'(누가 언제 뭘 했는지)를 최근 N줄만 보여주는 API.
try:
qs = parse_qs(parsed.query)
lines_raw = qs.get("lines", ["50"])[0].strip()
try:
lines = int(lines_raw)
except ValueError:
lines = 50
if lines < 1:
lines = 1
if lines > 300:
lines = 300
if not os.path.exists(AUDIT_LOG_PATH):
self._audit("audit_log", "-", 404, "error", "log_not_found")
self._send_json(404, {"status": "error", "message": "audit log not found", "error": "log_not_found"})
return
with open(AUDIT_LOG_PATH, "r", encoding="utf-8", errors="replace") as f:
content = f.readlines()
tail_lines = list(reversed([line.rstrip("\n") for line in content[-lines:]]))
self._audit("audit_log", "-", 200, "ok", f"lines={lines}")
self._send_json(200, {"status": "ok", "count": len(tail_lines), "lines": tail_lines})
except Exception as e:
self._audit("audit_log", "-", 500, "error", str(e))
self._send_json(500, {"status": "error", "message": "server error", "error": "server_error", "detail": str(e)})
def handle_dkim_list(self):
# 각 도메인에 등록된 DKIM 서명 키 목록을 보여주는 API.
try:
items = []
if os.path.isdir(DKIM_BASE_DIR):
for domain in sorted(os.listdir(DKIM_BASE_DIR)):
domain_dir = self._dkim_domain_dir(domain)
if not os.path.isdir(domain_dir):
continue
txt_files = sorted(glob.glob(os.path.join(domain_dir, "*.txt")))
private_files = sorted(glob.glob(os.path.join(domain_dir, "*.private")))
selectors = []
for txt_path in txt_files:
selector = os.path.basename(txt_path)[:-4]
selectors.append(
{
"selector": selector,
"txt_path": txt_path,
"private_key_path": self._dkim_private_key_path(domain, selector),
"txt_exists": os.path.exists(txt_path),
"private_exists": os.path.exists(self._dkim_private_key_path(domain, selector)),
}
)
items.append(
{
"domain": domain,
"path": domain_dir,
"selector_count": len(selectors),
"selectors": selectors,
"private_file_count": len(private_files),
}
)
self._audit("dkim_list", "-", 200, "ok", f"count={len(items)}")
self._send_json(200, {"status": "ok", "count": len(items), "domains": items})
except Exception as e:
self._audit("dkim_list", "-", 500, "error", str(e))
self._send_json(500, {"status": "error", "message": "server error", "error": "server_error", "detail": str(e)})
def handle_dkim_public(self, parsed):
# 특정 도메인의 DKIM 공개키(DNS에 등록해야 할 TXT 값)를 보여주는 API.
try:
qs = parse_qs(parsed.query)
domain = qs.get("domain", [""])[0].strip().lower()
selector = qs.get("selector", [DKIM_DEFAULT_SELECTOR])[0].strip()
if not domain:
self._audit("dkim_public", "-", 400, "error", "domain_required")
self._send_json(400, {"status": "error", "message": "domain is required", "error": "domain_required"})
return
txt_path = self._dkim_txt_path(domain, selector)
if not os.path.exists(txt_path):
self._audit("dkim_public", domain, 404, "error", "txt_not_found")
self._send_json(
404,
{
"status": "error",
"message": "dkim txt not found",
"error": "txt_not_found",
"domain": domain,
"selector": selector,
"txt_path": txt_path,
}
)
return
content = self._read_file_text(txt_path)
self._audit("dkim_public", domain, 200, "ok", f"selector={selector}")
self._send_json(
200,
{
"status": "ok",
"domain": domain,
"selector": selector,
"txt_path": txt_path,
"record": content,
}
)
except Exception as e:
self._audit("dkim_public", "-", 500, "error", str(e))
self._send_json(500, {"status": "error", "message": "server error", "error": "server_error", "detail": str(e)})
def handle_dkim_generate(self, data):
# 특정 도메인의 새 DKIM 키 쌍(개인키+공개키)을 생성하는 API.
domain = data.get("domain", "").strip().lower()
selector = data.get("selector", DKIM_DEFAULT_SELECTOR).strip()
force = self._normalize_active(data.get("force", 0))
bits = int(data.get("bits", DKIM_KEY_BITS))
try:
if not domain:
self._audit("dkim_generate", "-", 400, "error", "domain_required")
self._send_json(400, {"status": "error", "message": "domain is required", "error": "domain_required"})
return
if not selector:
self._audit("dkim_generate", domain, 400, "error", "selector_required")
self._send_json(400, {"status": "error", "message": "selector is required", "error": "selector_required"})
return
domain_dir = self._dkim_domain_dir(domain)
os.makedirs(domain_dir, exist_ok=True)
private_path = self._dkim_private_key_path(domain, selector)
txt_path = self._dkim_txt_path(domain, selector)
if (os.path.exists(private_path) or os.path.exists(txt_path)) and not force:
self._audit("dkim_generate", domain, 409, "error", "dkim_exists")
self._send_json(
409,
{
"status": "error",
"message": "dkim files already exist",
"error": "dkim_exists",
"domain": domain,
"selector": selector,
"private_key_path": private_path,
"txt_path": txt_path,
}
)
return
self._run(
[
"/usr/sbin/opendkim-genkey",
"-b", str(bits),
"-d", domain,
"-s", selector,
"-D", domain_dir,
]
)
default_private = os.path.join(domain_dir, f"{selector}.private")
default_txt = os.path.join(domain_dir, f"{selector}.txt")
if not os.path.exists(default_private) or not os.path.exists(default_txt):
raise RuntimeError("opendkim-genkey did not create expected files")
content = self._read_file_text(default_txt)
self._audit("dkim_generate", domain, 200, "ok", f"selector={selector},bits={bits}")
self._send_json(
200,
{
"status": "ok",
"message": "dkim generated",
"domain": domain,
"selector": selector,
"bits": bits,
"private_key_path": default_private,
"txt_path": default_txt,
"record": content,
}
)
except subprocess.CalledProcessError as e:
detail = e.stderr.strip() if e.stderr else str(e)
self._audit("dkim_generate", domain or "-", 500, "error", detail)
self._send_json(500, {"status": "error", "message": "dkim generate failed", "error": "dkim_generate_failed", "detail": detail})
except Exception as e:
self._audit("dkim_generate", domain or "-", 500, "error", str(e))
self._send_json(500, {"status": "error", "message": "server error", "error": "server_error", "detail": str(e)})
def handle_backup_list(self):
# 지금까지 만들어둔 백업 파일 목록을 보여주는 API.
try:
os.makedirs(BACKUP_BASE_DIR, exist_ok=True)
items = []
for name in sorted(os.listdir(BACKUP_BASE_DIR), reverse=True):
path = os.path.join(BACKUP_BASE_DIR, name)
if not os.path.isfile(path):
continue
st = os.stat(path)
items.append(
{
"file": name,
"path": path,
"size_bytes": st.st_size,
"size_human": self._human_size(st.st_size),
"mtime": datetime.fromtimestamp(st.st_mtime, tz=timezone.utc).isoformat(),
}
)
self._audit("backup_list", "-", 200, "ok", f"count={len(items)}")
self._send_json(200, {"status": "ok", "count": len(items), "backups": items})
except Exception as e:
self._audit("backup_list", "-", 500, "error", str(e))
self._send_json(500, {"status": "error", "message": "server error", "error": "server_error", "detail": str(e)})
def handle_backup_create(self, data):
# DB 전체와 메일/설정 폴더를 압축해서 백업 파일 하나로 만드는 API.
label = data.get("label", "").strip()
try:
os.makedirs(BACKUP_BASE_DIR, exist_ok=True)
ts = datetime.now().strftime("%Y%m%d-%H%M%S")
suffix = f"-{label}" if label else ""
backup_name = f"mailadmin-backup-{ts}{suffix}.tar.gz"
backup_path = os.path.join(BACKUP_BASE_DIR, backup_name)
with tempfile.TemporaryDirectory() as tmpdir:
sql_path = os.path.join(tmpdir, "mailserver.sql")
meta_path = os.path.join(tmpdir, "meta.json")
files_dir = os.path.join(tmpdir, "files")
os.makedirs(files_dir, exist_ok=True)
dump_cmd = [
MAILAPI_SQL_DUMP_BIN,
"-h", DB_CONFIG["host"],
"-u", DB_CONFIG["user"],
"--single-transaction",
"--skip-lock-tables",
DB_CONFIG["db"],
]
result = self._run(dump_cmd, env=self._mysql_env())
with open(sql_path, "w", encoding="utf-8") as f:
f.write(result.stdout)
for src in [MAIL_BASE_DIR, DKIM_BASE_DIR, "/etc/postfix", "/etc/dovecot"]:
if os.path.exists(src):
dst = os.path.join(files_dir, src.lstrip("/"))
os.makedirs(os.path.dirname(dst), exist_ok=True)
if os.path.isdir(src):
shutil.copytree(src, dst, symlinks=True, dirs_exist_ok=True)
else:
shutil.copy2(src, dst)
meta = {
"created_at": datetime.now(timezone.utc).isoformat(),
"db_name": DB_CONFIG["db"],
"mail_base_dir": MAIL_BASE_DIR,
"dkim_base_dir": DKIM_BASE_DIR,
"label": label,
}
with open(meta_path, "w", encoding="utf-8") as f:
json.dump(meta, f, ensure_ascii=False, indent=2)
self._run(["tar", "-C", tmpdir, "-czf", backup_path, "."])
st = os.stat(backup_path)
self._audit("backup_create", backup_name, 200, "ok", backup_path)
self._send_json(
200,
{
"status": "ok",
"message": "backup created",
"file": backup_name,
"path": backup_path,
"size_bytes": st.st_size,
"size_human": self._human_size(st.st_size),
}
)
except subprocess.CalledProcessError as e:
detail = e.stderr.strip() if e.stderr else str(e)
self._audit("backup_create", label or "-", 500, "error", detail)
self._send_json(500, {"status": "error", "message": "backup create failed", "error": "backup_create_failed", "detail": detail})
except Exception as e:
self._audit("backup_create", label or "-", 500, "error", str(e))
self._send_json(500, {"status": "error", "message": "server error", "error": "server_error", "detail": str(e)})
def handle_backup_restore(self, data):
# 백업 파일을 골라서 DB와 메일/설정 폴더를 그 시점 상태로 되돌리는 API. 잘못 쓰면 되돌릴 수 없으니 신중하게 다뤄야 한다.
# [보안 수정] file_name을 그대로 os.path.join에 넣으면
# "../../etc/something" 같은 값으로 백업 폴더 밖의 경로를
# 조작할 수 있었다. os.path.basename()으로 순수 파일명만
# 남겨서 경로 탈출(디렉터리 트래버설)을 원천 차단한다.
file_name = os.path.basename(data.get("file", "").strip())
try:
if not file_name:
self._audit("backup_restore", "-", 400, "error", "file_required")
self._send_json(400, {"status": "error", "message": "file is required", "error": "file_required"})
return
backup_path = os.path.join(BACKUP_BASE_DIR, file_name)
if not os.path.exists(backup_path):
self._audit("backup_restore", file_name, 404, "error", "backup_not_found")
self._send_json(404, {"status": "error", "message": "backup not found", "error": "backup_not_found", "file": file_name})
return
with tempfile.TemporaryDirectory() as tmpdir:
self._run(["tar", "-C", tmpdir, "-xzf", backup_path])
sql_path = os.path.join(tmpdir, "mailserver.sql")
files_dir = os.path.join(tmpdir, "files")
if os.path.exists(sql_path):
with open(sql_path, "r", encoding="utf-8", errors="replace") as f:
sql_content = f.read()
mysql_cmd = [
MAILAPI_SQL_BIN,
"-h", DB_CONFIG["host"],
"-u", DB_CONFIG["user"],
DB_CONFIG["db"],
]
subprocess.run(
mysql_cmd,
input=sql_content,
text=True,
env=self._mysql_env(),
capture_output=True,
check=True,
)
restore_targets = [
MAIL_BASE_DIR,
DKIM_BASE_DIR,
"/etc/postfix",
"/etc/dovecot",
]
for target in restore_targets:
source = os.path.join(files_dir, target.lstrip("/"))
if os.path.exists(source):
if os.path.isdir(source):
os.makedirs(target, exist_ok=True)
shutil.copytree(source, target, symlinks=True, dirs_exist_ok=True)
else:
os.makedirs(os.path.dirname(target), exist_ok=True)
shutil.copy2(source, target)
self._audit("backup_restore", file_name, 200, "ok", backup_path)
self._send_json(
200,
{
"status": "ok",
"message": "backup restored",
"file": file_name,
"path": backup_path,
}
)
except subprocess.CalledProcessError as e:
detail = e.stderr.strip() if e.stderr else str(e)
self._audit("backup_restore", file_name or "-", 500, "error", detail)
self._send_json(500, {"status": "error", "message": "backup restore failed", "error": "backup_restore_failed", "detail": detail})
except Exception as e:
self._audit("backup_restore", file_name or "-", 500, "error", str(e))
self._send_json(500, {"status": "error", "message": "server error", "error": "server_error", "detail": str(e)})
def handle_alias_list(self):
# 등록된 이메일 별칭(alias, 예: info@도메인 → 실제계정) 목록을 보여주는 API.
conn = None
cur = None
try:
conn = self._get_db()
cur = conn.cursor()
cur.execute(
"""
SELECT
a.id,
a.source,
a.destination,
a.active,
d.name AS domain
FROM virtual_aliases a
JOIN virtual_domains d ON a.domain_id = d.id
ORDER BY a.source ASC
"""
)
rows = cur.fetchall()
items = []
for row in rows:
items.append(
{
"id": row[0],
"source": row[1],
"destination": row[2],
"active": int(row[3]),
"domain": row[4],
}
)
self._audit("alias_list", "-", 200, "ok", f"count={len(items)}")
self._send_json(200, {"status": "ok", "count": len(items), "aliases": items})
except Exception as e:
self._audit("alias_list", "-", 500, "error", str(e))
self._send_json(500, {"status": "error", "message": "server error", "error": "server_error", "detail": str(e)})
finally:
if cur:
cur.close()
if conn:
conn.close()
def handle_alias_add(self, data):
# 이메일 별칭을 새로 등록하는 API.
conn = None
cur = None
source = data.get("source", "").strip().lower()
destination = data.get("destination", "").strip().lower()
try:
active = self._normalize_active(data.get("active", 1))
if not source:
self._audit("alias_add", "-", 400, "error", "source_required")
self._send_json(400, {"status": "error", "message": "source is required", "error": "source_required"})
return
if not destination:
self._audit("alias_add", source, 400, "error", "destination_required")
self._send_json(400, {"status": "error", "message": "destination is required", "error": "destination_required"})
return
_, source_domain = self._split_email(source)
if not source_domain:
self._audit("alias_add", source, 400, "error", "invalid_source")
self._send_json(400, {"status": "error", "message": "invalid source format (must be local@domain)", "error": "invalid_source", "source": source})
return
conn = self._get_db()
cur = conn.cursor()
cur.execute("SELECT id FROM virtual_domains WHERE name=%s", (source_domain,))
domain_row = cur.fetchone()
if not domain_row:
self._audit("alias_add", source, 404, "error", "domain_not_found")
self._send_json(404, {"status": "error", "message": "domain not found", "error": "domain_not_found", "domain": source_domain})
return
domain_id = domain_row[0]
cur.execute("SELECT id FROM virtual_aliases WHERE source=%s AND destination=%s", (source, destination))
if cur.fetchone():
self._audit("alias_add", source, 409, "error", "alias_exists")
self._send_json(409, {"status": "error", "message": "alias already exists", "error": "alias_exists", "source": source, "destination": destination})
return
cur.execute(
"INSERT INTO virtual_aliases (domain_id, source, destination, active) VALUES (%s, %s, %s, %s)",
(domain_id, source, destination, active)
)
conn.commit()
self._audit("alias_add", source, 200, "ok", f"destination={destination},active={active}")
self._send_json(
200,
{
"status": "ok",
"message": "alias added",
"source": source,
"destination": destination,
"domain": source_domain,
"active": active,
}
)
except Exception as e:
self._audit("alias_add", source or "-", 500, "error", str(e))
self._send_json(500, {"status": "error", "message": "server error", "error": "server_error", "detail": str(e)})
finally:
if cur:
cur.close()
if conn:
conn.close()
def handle_alias_delete(self, data):
# 이메일 별칭을 삭제하는 API.
conn = None
cur = None
alias_id = data.get("id")
try:
if not alias_id:
self._audit("alias_delete", "-", 400, "error", "id_required")
self._send_json(400, {"status": "error", "message": "id is required", "error": "id_required"})
return
conn = self._get_db()
cur = conn.cursor()
cur.execute("DELETE FROM virtual_aliases WHERE id=%s", (alias_id,))
if cur.rowcount == 0:
self._audit("alias_delete", str(alias_id), 404, "error", "alias_not_found")
self._send_json(404, {"status": "error", "message": "alias not found", "error": "alias_not_found", "id": alias_id})
return
conn.commit()
self._audit("alias_delete", str(alias_id), 200, "ok", "alias_deleted")
self._send_json(200, {"status": "ok", "message": "alias deleted", "id": alias_id})
except Exception as e:
self._audit("alias_delete", str(alias_id) if alias_id else "-", 500, "error", str(e))
self._send_json(500, {"status": "error", "message": "server error", "error": "server_error", "detail": str(e)})
finally:
if cur:
cur.close()
if conn:
conn.close()
def handle_alias_set_active(self, data):
# 이메일 별칭을 켜고 끄는 API.
conn = None
cur = None
alias_id = data.get("id")
try:
active = self._normalize_active(data.get("active", 0))
if not alias_id:
self._audit("alias_set_active", "-", 400, "error", "id_required")
self._send_json(400, {"status": "error", "message": "id is required", "error": "id_required"})
return
conn = self._get_db()
cur = conn.cursor()
cur.execute("UPDATE virtual_aliases SET active=%s WHERE id=%s", (active, alias_id))
if cur.rowcount == 0:
self._audit("alias_set_active", str(alias_id), 404, "error", "alias_not_found")
self._send_json(404, {"status": "error", "message": "alias not found", "error": "alias_not_found", "id": alias_id})
return
conn.commit()
self._audit("alias_set_active", str(alias_id), 200, "ok", f"active={active}")
self._send_json(200, {"status": "ok", "message": "alias active updated", "id": alias_id, "active": active})
except Exception as e:
self._audit("alias_set_active", str(alias_id) if alias_id else "-", 500, "error", str(e))
self._send_json(500, {"status": "error", "message": "server error", "error": "server_error", "detail": str(e)})
finally:
if cur:
cur.close()
if conn:
conn.close()
def handle_system_status(self):
# postfix, dovecot, mariadb 같은 핵심 서비스가 지금 살아있는지(active) 확인하는 API.
try:
items = []
for service in SYSTEMD_SERVICES:
try:
result = subprocess.run(
["systemctl", "is-active", service],
capture_output=True,
text=True,
check=False,
)
state = (result.stdout or "").strip() or "unknown"
except FileNotFoundError:
state = "unknown"
items.append(
{
"service": service,
"active": state == "active",
"state": state,
}
)
self._audit("system_status", "-", 200, "ok", f"count={len(items)}")
self._send_json(200, {"status": "ok", "count": len(items), "services": items})
except Exception as e:
self._audit("system_status", "-", 500, "error", str(e))
self._send_json(500, {"status": "error", "message": "server error", "error": "server_error", "detail": str(e)})
def handle_fail2ban_status(self):
# fail2ban(반복 로그인 실패 시 자동 차단하는 프로그램)이 지금 어떤 IP를 차단 중인지 보여주는 API.
try:
jail_result = subprocess.run(
["fail2ban-client", "status"],
capture_output=True,
text=True,
check=False,
)
if jail_result.returncode != 0:
self._audit("fail2ban_status", "-", 200, "ok", "fail2ban_unavailable")
self._send_json(200, {"status": "ok", "available": False, "jails": []})
return
jail_names = []
for line in jail_result.stdout.splitlines():
if "Jail list:" in line:
raw = line.split("Jail list:", 1)[1].strip()
jail_names = [j.strip() for j in raw.split(",") if j.strip()]
break
jails = []
for name in jail_names:
detail = subprocess.run(
["fail2ban-client", "status", name],
capture_output=True,
text=True,
check=False,
)
banned_ips = []
currently_banned = 0
for line in detail.stdout.splitlines():
line = line.strip()
line = re.sub(r'^[|`\-\s]+', '', line) # fail2ban 트리 구조 문자(|-, `-) 제거
if ":" not in line:
continue
key, _, val = line.partition(":")
key = key.strip()
val = val.strip()
if key == "Currently banned":
try:
currently_banned = int(val)
except ValueError:
currently_banned = 0
elif key == "Banned IP list":
banned_ips = [ip for ip in val.split() if ip]
jails.append(
{
"jail": name,
"currently_banned": currently_banned,
"banned_ips": banned_ips,
}
)
total_banned = sum(j["currently_banned"] for j in jails)
self._audit("fail2ban_status", "-", 200, "ok", f"jails={len(jails)},banned={total_banned}")
self._send_json(
200,
{
"status": "ok",
"available": True,
"total_banned": total_banned,
"jails": jails,
}
)
except FileNotFoundError:
self._audit("fail2ban_status", "-", 200, "ok", "fail2ban_not_installed")
self._send_json(200, {"status": "ok", "available": False, "jails": []})
except Exception as e:
self._audit("fail2ban_status", "-", 500, "error", str(e))
self._send_json(500, {"status": "error", "message": "server error", "error": "server_error", "detail": str(e)})
def handle_fail2ban_unban(self, data):
# fail2ban이 차단한 특정 IP를 수동으로 풀어주는 API.
ip = (data.get("ip") or "").strip()
jail = (data.get("jail") or "").strip()
try:
if not ip:
self._audit("fail2ban_unban", "-", 400, "error", "ip_required")
self._send_json(400, {"status": "error", "message": "ip is required", "error": "ip_required"})
return
# 간단한 형식 검증 (IPv4/IPv6 문자만 허용, 명령 인젝션 방지)
if not re.match(r'^[0-9a-fA-F:.]+$', ip):
self._audit("fail2ban_unban", ip, 400, "error", "invalid_ip")
self._send_json(400, {"status": "error", "message": "invalid ip format", "error": "invalid_ip"})
return
if jail:
jail_names = [jail]
else:
jail_result = subprocess.run(
["fail2ban-client", "status"],
capture_output=True,
text=True,
check=False,
)
jail_names = []
for line in jail_result.stdout.splitlines():
if "Jail list:" in line:
raw = line.split("Jail list:", 1)[1].strip()
jail_names = [j.strip() for j in raw.split(",") if j.strip()]
break
results = []
any_success = False
for name in jail_names:
r = subprocess.run(
["fail2ban-client", "set", name, "unbanip", ip],
capture_output=True,
text=True,
check=False,
)
ok = (r.returncode == 0)
if ok:
any_success = True
results.append(
{
"jail": name,
"ok": ok,
"detail": (r.stdout or r.stderr or "").strip(),
}
)
self._audit("fail2ban_unban", ip, 200, "ok" if any_success else "error", f"jails={jail_names}")
self._send_json(
200,
{
"status": "ok" if any_success else "error",
"ip": ip,
"results": results,
}
)
except Exception as e:
self._audit("fail2ban_unban", ip or "-", 500, "error", str(e))
self._send_json(500, {"status": "error", "message": "server error", "error": "server_error", "detail": str(e)})
def handle_cert_status(self):
# Let's Encrypt 인증서가 도메인별로 언제 만료되는지 계산해서 보여주는 API.
try:
items = []
pattern = os.path.join(CERT_LIVE_DIR, "*", "cert.pem")
for cert_path in sorted(glob.glob(pattern)):
domain = os.path.basename(os.path.dirname(cert_path))
try:
result = subprocess.run(
["openssl", "x509", "-enddate", "-noout", "-in", cert_path],
capture_output=True,
text=True,
check=True,
)
raw = result.stdout.strip()
end_str = raw.split("=", 1)[1].strip() if "=" in raw else ""
expires_at = datetime.strptime(end_str, "%b %d %H:%M:%S %Y %Z").replace(tzinfo=timezone.utc)
days_remaining = (expires_at - datetime.now(timezone.utc)).days
items.append(
{
"domain": domain,
"path": cert_path,
"expires_at": expires_at.isoformat(),
"days_remaining": days_remaining,
}
)
except Exception as inner_e:
items.append(
{
"domain": domain,
"path": cert_path,
"error": str(inner_e),
}
)
self._audit("cert_status", "-", 200, "ok", f"count={len(items)}")
self._send_json(200, {"status": "ok", "count": len(items), "certs": items})
except Exception as e:
self._audit("cert_status", "-", 500, "error", str(e))
self._send_json(500, {"status": "error", "message": "server error", "error": "server_error", "detail": str(e)})
def handle_tempmail_stats(self):
# 임시메일 기능의 현재 사용 현황(활성 계정 수, 오늘 생성 수, 최근 어뷰징 시도 수)을 보여주는 API.
conn = None
cur = None
try:
conn = self._get_db()
cur = conn.cursor()
cur.execute("SELECT COUNT(*) FROM tempmail_accounts")
active_count = cur.fetchone()[0]
cur.execute("SELECT COUNT(*) FROM tempmail_creation_log WHERE created_at > NOW() - INTERVAL 1 DAY")
created_today = cur.fetchone()[0]
# 최근 1시간 내 어뷰징 로그 라인 수
abuse_count_1h = 0
abuse_path = LOG_SOURCES.get("tempmail_abuse", {}).get("path", "")
if abuse_path and os.path.isfile(abuse_path):
one_hour_ago = datetime.now() - timedelta(hours=1)
size = os.path.getsize(abuse_path)
read_bytes = min(size, LOG_TAIL_MAX_BYTES)
with open(abuse_path, "rb") as f:
f.seek(max(0, size - read_bytes))
data = f.read()
text = data.decode("utf-8", errors="replace")
for line in text.splitlines():
m = re.match(r"^\[(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2})\]", line)
if not m:
continue
try:
ts = datetime.strptime(m.group(1), "%Y-%m-%d %H:%M:%S")
except ValueError:
continue
if ts >= one_hour_ago:
abuse_count_1h += 1
result = {
"status": "ok",
"active_accounts": int(active_count),
"created_today": int(created_today),
"abuse_count_1h": abuse_count_1h,
}
self._audit("tempmail_stats", "-", 200, "ok", f"active={active_count},abuse_1h={abuse_count_1h}")
self._send_json(200, result)
except Exception as e:
self._audit("tempmail_stats", "-", 500, "error", str(e))
self._send_json(500, {"status": "error", "message": "server error", "error": "server_error", "detail": str(e)})
finally:
if cur:
cur.close()
if conn:
conn.close()
def _grep_tail_file(self, path, patterns=None, max_lines=50, max_read_bytes=2 * 1024 * 1024, reverse=True):
# 로그 파일 끝부분만 읽어서, 필요하면 특정 패턴에 맞는 줄만 골라내는 공용 함수.
if not os.path.isfile(path):
return [f"(파일 없음: {path})"]
try:
size = os.path.getsize(path)
read_bytes = min(size, max_read_bytes)
with open(path, "rb") as f:
f.seek(max(0, size - read_bytes))
data = f.read()
text = data.decode("utf-8", errors="replace")
lines = text.splitlines()
if patterns:
lines = [l for l in lines if any(re.search(p, l, re.IGNORECASE) for p in patterns)]
result = lines[-max_lines:] if lines else ["(해당 없음)"]
if reverse and result != ["(해당 없음)"]:
result = list(reversed(result))
return result
except PermissionError:
return ["(읽기 권한 없음)"]
except Exception as e:
return [f"(읽기 실패: {e})"]
def _run_cmd_lines(self, cmd, max_lines=50, reverse=False, already_sorted_desc=False):
# 리눅스 명령어를 실행하고 그 출력을 줄 단위로 잘라서 돌려주는 공용 함수. last/lastb처럼 '이미 최신순 정렬'된 명령은 앞에서부터, 그 외에는 뒤에서부터 자른다.
try:
result = subprocess.run(cmd, capture_output=True, text=True, check=False, timeout=15)
out = (result.stdout or "").strip()
err = (result.stderr or "").strip()
if not out and err:
return [f"(명령 실패: {err[:300]})"]
lines = out.splitlines()
if already_sorted_desc:
# last/lastb는 이미 최신순으로 출력되지만, 맨 끝에 빈 줄 + "wtmp/btmp begins ..."
# 안내줄이 붙어서 나온다. 뒤에서 max_lines개를 자르면 정작 맨 위(=가장 최신) 몇 건이
# 잘려나가고 의미 없는 안내줄만 남는 문제가 있었음 — 앞에서부터 잘라야 최신 기록이 보존됨.
filtered = [
l for l in lines
if l.strip() and not re.match(r'^(wtmp|btmp) begins', l.strip())
]
capped = filtered[:max_lines] if filtered else ["(해당 없음)"]
else:
capped = lines[-max_lines:] if lines else ["(해당 없음)"]
if reverse and capped != ["(해당 없음)"]:
capped = list(reversed(capped))
return capped
except FileNotFoundError:
return [f"(명령 없음: {cmd[0]})"]
except subprocess.TimeoutExpired:
return ["(명령 시간 초과)"]
except Exception as e:
return [f"(실행 실패: {e})"]
def handle_security_checks(self, parsed):
# SSH 로그인 이력, fail2ban 차단, 크론잡, 실행 중인 서비스 등 서버 보안 상태를 한 화면에서 훑어볼 수 있게 모아서 보여주는 API.
qs = parse_qs(parsed.query)
lines_raw = qs.get("lines", ["50"])[0]
try:
lines = int(lines_raw)
except ValueError:
lines = 50
lines = max(1, min(lines, 500))
try:
items = []
items.append({
"key": "ssh_success",
"label": f"SSH 로그인 성공 이력 (최근 {lines}건)",
"hint": "낯선 IP나 본인이 접속하지 않은 시간대가 있으면 침해 의심 신호입니다.",
"lines": self._run_cmd_lines(["last", "-a", "-n", str(lines)], max_lines=lines, already_sorted_desc=True),
})
items.append({
"key": "ssh_failed",
"label": f"SSH 로그인 실패 이력 (최근 {lines}건)",
"hint": "반복 실패는 흔한 브루트포스 시도입니다. fail2ban이 정상 차단 중이면 큰 문제는 아닙니다.",
"lines": self._run_cmd_lines(["lastb", "-n", str(lines)], max_lines=lines, already_sorted_desc=True),
})
items.append({
"key": "fail2ban_bans",
"label": f"fail2ban 최근 차단 이력 (최근 {lines}건)",
"hint": "평소보다 밴 빈도가 급증했으면 본격적인 스캐닝/공격이 진행 중이라는 신호입니다.",
"lines": self._grep_tail_file("/var/log/fail2ban.log", patterns=[r"\bBan\b"], max_lines=lines),
})
items.append({
"key": "mail_sasl_fail",
"label": f"메일 SMTP 인증 실패 (최근 {lines}건)",
"hint": "SASL 인증 실패가 갑자기 몰리면 계정 탈취를 노린 시도일 수 있습니다.",
"lines": self._grep_tail_file("/var/log/mail.log", patterns=[r"sasl", r"authentication failed"], max_lines=lines),
})
sent_count_lines = self._grep_tail_file("/var/log/mail.log", patterns=[r"status=sent"], max_lines=100000)
sent_count = 0 if sent_count_lines == ["(해당 없음)"] else len(sent_count_lines)
items.append({
"key": "mail_sent_count",
"label": "메일 발송 성공 건수 (현재 로그 범위 내)",
"hint": "평소 대비 비정상적으로 많으면, 어떤 계정이 뚫려 스팸 릴레이로 악용되고 있을 수 있습니다.",
"lines": [f"총 {sent_count}건"],
})
items.append({
"key": "tempmail_abuse",
"label": f"임시메일 어뷰징 로그 (최근 {lines}건)",
"hint": "invalid_recovery_token이 짧은 시간에 대량이면 토큰 무차별 대입 시도입니다.",
"lines": self._grep_tail_file(
LOG_SOURCES.get("tempmail_abuse", {}).get("path", "/var/log/tempmail/abuse.log"),
max_lines=lines,
),
})
tempmail_access_path = LOG_SOURCES.get("tempmail_access", {}).get("path", "/var/log/nginx/tempmail-access.log")
top404 = []
try:
if os.path.isfile(tempmail_access_path):
counts = {}
with open(tempmail_access_path, "r", encoding="utf-8", errors="replace") as f:
for line in f:
m = re.search(r'"[A-Z]+ (\S+)[^"]*"\s+404\s', line)
if m:
path = m.group(1)
counts[path] = counts.get(path, 0) + 1
top = sorted(counts.items(), key=lambda x: x[1], reverse=True)[:20]
top404 = [f"{cnt}회 {p}" for p, cnt in top] if top else ["(404 없음)"]
else:
top404 = ["(파일 없음)"]
except Exception as e:
top404 = [f"(집계 실패: {e})"]
items.append({
"key": "tempmail_404_top",
"label": "임시메일 사이트 404 상위 경로 (상위 20개)",
"hint": "존재하지 않는 경로(/wp-admin, /.env 등)를 다수 두드리는 건 흔한 자동 스캐너입니다.",
"lines": top404,
})
tempmail_error_path = LOG_SOURCES.get("tempmail_error", {}).get("path", "/var/log/nginx/tempmail-error.log")
items.append({
"key": "tempmail_5xx",
"label": f"임시메일 사이트 5xx 에러 (최근 {lines}건)",
"hint": "비정상적으로 에러가 몰리면 부하 공격이나 앱 취약점을 건드리고 있다는 신호입니다.",
"lines": self._grep_tail_file(tempmail_error_path, patterns=[r"\b(500|502|503)\b"], max_lines=lines),
})
mailadmin_access_path = LOG_SOURCES.get("php_ui_journal", {}).get("path", "/var/log/nginx/mailadmin-access.log")
mailadmin_top404 = []
try:
if os.path.isfile(mailadmin_access_path):
counts = {}
with open(mailadmin_access_path, "r", encoding="utf-8", errors="replace") as f:
for line in f:
m = re.search(r'"[A-Z]+ (\S+)[^"]*"\s+404\s', line)
if m:
path = m.group(1)
counts[path] = counts.get(path, 0) + 1
top = sorted(counts.items(), key=lambda x: x[1], reverse=True)[:20]
mailadmin_top404 = [f"{cnt}회 {p}" for p, cnt in top] if top else ["(404 없음)"]
else:
mailadmin_top404 = ["(파일 없음)"]
except Exception as e:
mailadmin_top404 = [f"(집계 실패: {e})"]
items.append({
"key": "mailadmin_404_top",
"label": "Mail Admin UI 404 상위 경로 (상위 20개)",
"hint": "SSH 터널로만 접근 가능한 대시보드라 위험도는 낮지만, 존재하지 않는 경로를 다수 두드리면 확인이 필요합니다.",
"lines": mailadmin_top404,
})
mailadmin_error_path = LOG_SOURCES.get("mailadmin_error", {}).get("path", "/var/log/nginx/mailadmin-error.log")
items.append({
"key": "mailadmin_5xx",
"label": f"Mail Admin UI 5xx 에러 (최근 {lines}건)",
"hint": "비정상적으로 에러가 몰리면 php-fpm 풀 문제나 코드 버그를 의심해볼 신호입니다.",
"lines": self._grep_tail_file(mailadmin_error_path, patterns=[r"\b(500|502|503)\b"], max_lines=lines),
})
items.append({
"key": "dovecot_errors",
"label": f"Dovecot 자체 에러 로그 (최근 {lines}건)",
"hint": "비정상 로그인 패턴이나 dovecot 자체 오류를 확인합니다.",
"lines": self._run_cmd_lines(["doveadm", "log", "errors"], max_lines=lines, reverse=True),
})
items.append({
"key": "cron_dirs",
"label": "/etc/cron.d, /etc/cron.daily 목록",
"hint": "본인이 넣지 않은 스크립트/심볼릭 링크가 있으면 침해 흔적일 수 있습니다.",
"lines": (
["--- /etc/cron.d ---"] + self._run_cmd_lines(["ls", "-la", "/etc/cron.d/"], max_lines=lines)
+ ["--- /etc/cron.daily ---"] + self._run_cmd_lines(["ls", "-la", "/etc/cron.daily/"], max_lines=lines)
),
})
crontab_lines = []
for cron_user in ["root", "www-data", "tempmail"]:
r = subprocess.run(["crontab", "-u", cron_user, "-l"], capture_output=True, text=True, check=False)
out = (r.stdout or "").strip()
if out:
crontab_lines.append(f"--- {cron_user} ---")
crontab_lines.extend(out.splitlines())
else:
crontab_lines.append(f"--- {cron_user}: 등록된 크론잡 없음 ---")
items.append({
"key": "crontabs",
"label": "등록된 크론잡 (root / www-data / tempmail)",
"hint": "본인이 등록하지 않은 낯선 크론잡이 있으면 침해 가능성이 매우 높습니다.",
"lines": crontab_lines,
})
items.append({
"key": "running_services",
"label": f"현재 실행 중인 systemd 서비스 (최대 {lines}건)",
"hint": "본인이 만들지 않은 낯선 서비스가 떠 있으면 침해 흔적일 수 있습니다.",
"lines": self._run_cmd_lines(
["systemctl", "list-units", "--type=service", "--state=running", "--no-legend", "--no-pager"],
max_lines=lines,
),
})
items.append({
"key": "recent_modified_files",
"label": f"최근 7일 내 수정된 시스템 파일 (최대 {lines}건)",
"hint": "본인이 최근에 안 건드린 /etc, /usr/local/bin, /opt 안의 파일이 있으면 확인이 필요합니다.",
"lines": self._run_cmd_lines(
["find", "/etc", "/usr/local/bin", "/opt", "-mtime", "-7", "-type", "f"],
max_lines=lines,
),
})
items.append({
"key": "established_connections",
"label": f"현재 연결된 세션 (ESTABLISHED, 최대 {lines}건)",
"hint": "낯선 외부 IP와 지속적으로 연결돼 있으면 백도어/역방향 셸 가능성을 의심해야 합니다.",
"lines": self._run_cmd_lines(["ss", "-tnp", "state", "established"], max_lines=lines),
})
self._audit("security_checks", "-", 200, "ok", f"count={len(items)},lines={lines}")
self._send_json(200, {"status": "ok", "count": len(items), "lines": lines, "checks": items})
except Exception as e:
self._audit("security_checks", "-", 500, "error", str(e))
self._send_json(500, {"status": "error", "message": "server error", "error": "server_error", "detail": str(e)})
def handle_log_sources(self):
# 대시보드의 '서버 로그' 탭에서 조회 가능한 로그 종류 목록(파일 경로, 존재 여부, 크기)을 보여주는 API.
try:
items = []
for key, conf in LOG_SOURCES.items():
if conf.get("type") == "journal":
items.append(
{
"key": key,
"label": conf["label"],
"path": f"journalctl -u {conf['unit']}",
"exists": True,
"size_bytes": None,
"size_human": None,
}
)
else:
path = conf["path"]
exists = os.path.isfile(path)
size = os.path.getsize(path) if exists else 0
items.append(
{
"key": key,
"label": conf["label"],
"path": path,
"exists": exists,
"size_bytes": size if exists else None,
"size_human": self._human_size(size) if exists else None,
}
)
self._audit("log_sources", "-", 200, "ok", f"count={len(items)}")
self._send_json(200, {"status": "ok", "count": len(items), "sources": items})
except Exception as e:
self._audit("log_sources", "-", 500, "error", str(e))
self._send_json(500, {"status": "error", "message": "server error", "error": "server_error", "detail": str(e)})
def handle_log_tail(self, parsed):
# 특정 로그 파일(또는 journalctl)의 최근 N줄을 읽어서 보여주는 API.
qs = parse_qs(parsed.query)
key = qs.get("key", [""])[0].strip()
lines_raw = qs.get("lines", ["100"])[0]
try:
lines = int(lines_raw)
except ValueError:
lines = 100
lines = max(1, min(lines, 500))
try:
if key not in LOG_SOURCES:
self._audit("log_tail", key or "-", 400, "error", "unknown_log")
self._send_json(400, {"status": "error", "message": "unknown log key", "error": "unknown_log"})
return
conf = LOG_SOURCES[key]
if conf.get("type") == "journal":
result = subprocess.run(
["journalctl", "-u", conf["unit"], "-n", str(lines), "--no-pager", "-o", "short-iso"],
capture_output=True,
text=True,
check=False,
)
if result.returncode != 0:
self._audit("log_tail", key, 500, "error", "journalctl_failed")
self._send_json(500, {"status": "error", "message": "journalctl failed", "error": "journalctl_failed", "detail": result.stderr.strip()})
return
tail_lines = list(reversed(result.stdout.splitlines()))
self._audit("log_tail", key, 200, "ok", f"lines={len(tail_lines)}")
self._send_json(
200,
{
"status": "ok",
"key": key,
"label": conf["label"],
"path": f"journalctl -u {conf['unit']}",
"size_bytes": None,
"count": len(tail_lines),
"lines": tail_lines,
}
)
return
path = conf["path"]
if not os.path.isfile(path):
self._audit("log_tail", key, 404, "error", "log_not_found")
self._send_json(404, {"status": "error", "message": "log file not found", "error": "log_not_found", "path": path})
return
size = os.path.getsize(path)
read_bytes = min(size, LOG_TAIL_MAX_BYTES)
with open(path, "rb") as f:
f.seek(max(0, size - read_bytes))
data = f.read()
text = data.decode("utf-8", errors="replace")
all_lines = text.splitlines()
tail_lines = list(reversed(all_lines[-lines:]))
self._audit("log_tail", key, 200, "ok", f"lines={len(tail_lines)}")
self._send_json(
200,
{
"status": "ok",
"key": key,
"label": conf["label"],
"path": path,
"size_bytes": size,
"count": len(tail_lines),
"lines": tail_lines,
}
)
except PermissionError:
self._audit("log_tail", key, 403, "error", "permission_denied")
self._send_json(403, {"status": "error", "message": "permission denied reading this log", "error": "permission_denied"})
except Exception as e:
self._audit("log_tail", key or "-", 500, "error", str(e))
self._send_json(500, {"status": "error", "message": "server error", "error": "server_error", "detail": str(e)})
def handle_file_roots(self):
# 파일탐색기의 바로가기 목록과 접근 금지 경로 목록을 보여주는 API.
try:
self._audit("file_roots", "-", 200, "ok", f"count={len(FILE_MANAGER_QUICKLINKS)}")
self._send_json(
200,
{
"status": "ok",
"roots": FILE_MANAGER_QUICKLINKS,
"blocked_paths": FILE_MANAGER_BLOCKED_PATHS,
}
)
except Exception as e:
self._audit("file_roots", "-", 500, "error", str(e))
self._send_json(500, {"status": "error", "message": "server error", "error": "server_error", "detail": str(e)})
def handle_file_list(self, parsed):
# 파일탐색기에서 특정 폴더 안의 파일/폴더 목록을 보여주는 API.
qs = parse_qs(parsed.query)
rel_path = qs.get("path", ["/"])[0]
target = rel_path or "/"
try:
candidate = self._fm_resolve(rel_path)
if not os.path.isdir(candidate):
self._audit("file_list", target, 404, "error", "dir_not_found")
self._send_json(404, {"status": "error", "message": "directory not found", "error": "dir_not_found"})
return
entries = []
with os.scandir(candidate) as it:
for entry in it:
try:
entry_path = os.path.join(candidate, entry.name)
is_blocked = False
try:
self._fm_resolve(entry_path)
except ValueError:
is_blocked = True
st = entry.stat(follow_symlinks=False)
is_dir = entry.is_dir(follow_symlinks=False)
entries.append(
{
"name": entry.name,
"is_dir": is_dir,
"blocked": is_blocked,
"size_bytes": None if (is_dir or is_blocked) else st.st_size,
"size_human": None if (is_dir or is_blocked) else self._human_size(st.st_size),
"mtime": datetime.fromtimestamp(st.st_mtime, tz=timezone.utc).isoformat(),
}
)
except OSError:
continue
entries.sort(key=lambda e: (not e["is_dir"], e["name"].lower()))
rel_display = candidate if candidate != "/" else ""
self._audit("file_list", target, 200, "ok", f"count={len(entries)}")
self._send_json(
200,
{
"status": "ok",
"path": rel_display,
"count": len(entries),
"entries": entries,
}
)
except ValueError as e:
self._audit("file_list", target, 400, "error", str(e))
self._send_json(400, {"status": "error", "message": "이 경로는 접근이 차단되어 있습니다 (시스템 영역)", "error": str(e)})
except Exception as e:
self._audit("file_list", target, 500, "error", str(e))
self._send_json(500, {"status": "error", "message": "server error", "error": "server_error", "detail": str(e)})
def handle_file_read(self, parsed):
# 파일탐색기에서 특정 파일의 내용을 읽어서 보여주는 API. 너무 큰 파일은 거부한다.
qs = parse_qs(parsed.query)
rel_path = qs.get("path", [""])[0]
target = rel_path or "/"
try:
candidate = self._fm_resolve(rel_path)
if not os.path.isfile(candidate):
self._audit("file_read", target, 404, "error", "file_not_found")
self._send_json(404, {"status": "error", "message": "file not found", "error": "file_not_found"})
return
size = os.path.getsize(candidate)
if size > FILE_MANAGER_MAX_VIEW_BYTES:
self._audit("file_read", target, 413, "error", "file_too_large_to_view")
self._send_json(
413,
{
"status": "error",
"message": "file too large to view here",
"error": "file_too_large_to_view",
"size_bytes": size,
"limit_bytes": FILE_MANAGER_MAX_VIEW_BYTES,
}
)
return
content = self._read_file_text(candidate)
editable = size <= FILE_MANAGER_MAX_EDIT_BYTES
self._audit("file_read", target, 200, "ok", f"size={size},editable={editable}")
self._send_json(
200,
{
"status": "ok",
"path": candidate,
"size_bytes": size,
"editable": editable,
"content": content,
}
)
except ValueError as e:
self._audit("file_read", target, 400, "error", str(e))
self._send_json(400, {"status": "error", "message": "이 경로는 접근이 차단되어 있습니다 (시스템 영역)", "error": str(e)})
except Exception as e:
self._audit("file_read", target, 500, "error", str(e))
self._send_json(500, {"status": "error", "message": "server error", "error": "server_error", "detail": str(e)})
def handle_file_write(self, data):
# 파일탐색기에서 파일을 수정 저장하는 API. 저장 전에 자동으로 백업(.bak-타임스탬프)을 남긴다.
rel_path = data.get("path") or ""
content = data.get("content", "")
target = rel_path or "/"
try:
candidate = self._fm_resolve(rel_path)
if not os.path.isfile(candidate):
self._audit("file_write", target, 404, "error", "file_not_found")
self._send_json(
404,
{
"status": "error",
"message": "file not found (only existing files can be edited, not created)",
"error": "file_not_found",
}
)
return
current_size = os.path.getsize(candidate)
if current_size > FILE_MANAGER_MAX_EDIT_BYTES:
self._audit("file_write", target, 413, "error", "file_too_large_to_edit")
self._send_json(
413,
{
"status": "error",
"message": "file too large to edit (view-only)",
"error": "file_too_large_to_edit",
"size_bytes": current_size,
"limit_bytes": FILE_MANAGER_MAX_EDIT_BYTES,
}
)
return
ts = datetime.now().strftime("%Y%m%d-%H%M%S")
backup_path = f"{candidate}.bak-{ts}"
shutil.copy2(candidate, backup_path)
with open(candidate, "w", encoding="utf-8") as f:
f.write(content)
self._audit("file_write", target, 200, "ok", f"backup={backup_path}")
self._send_json(
200,
{
"status": "ok",
"message": "file saved",
"path": candidate,
"backup_path": backup_path,
}
)
except ValueError as e:
self._audit("file_write", target, 400, "error", str(e))
self._send_json(400, {"status": "error", "message": "이 경로는 접근이 차단되어 있습니다 (시스템 영역)", "error": str(e)})
except Exception as e:
self._audit("file_write", target, 500, "error", str(e))
self._send_json(500, {"status": "error", "message": "server error", "error": "server_error", "detail": str(e)})
if __name__ == "__main__":
print(f"mailadmin api listening on {API_BIND_HOST}:{API_BIND_PORT}")
print(f"audit log: {AUDIT_LOG_PATH}")
print(f"backup dir: {BACKUP_BASE_DIR}")
if not API_KEY:
print("WARNING: MAILAPI_API_KEY is empty - all requests will be rejected")
server = ThreadingHTTPServer((API_BIND_HOST, API_BIND_PORT), MailAdminHandler)
server.serve_forever()
간단히 구조만 짚어보면:
do_GET/do_POST: 요청이 들어오면 URL을 보고 어떤 처리 함수로 넘길지 정해주는 교통정리 역할을 합니다.handle_*함수들: 실제로 도메인 추가, 계정 삭제, 백업, 보안 점검 등 각 기능을 처리하는 함수들입니다._로 시작하는 함수들: 다른 함수들이 공통으로 쓰는 내부 도우미 함수입니다.
혹시 코드 보다가 궁금한 부분 있으면 댓글로 남겨주세요.