네트워크 장비나 보안솔루션 등 이벤트나 정기적인 리포트를 메일로 발송해주는 기능은 기본적으로 존재합니다.
하지만, 메일을 자주 사용하지 않은 경우 놓치거나 잘 활용하기 어려운 경우가 많습니다.
메일서버 없이 메일을 수신받고 해당 정보를 내부적으로 운영중인 통합 모니터링 시스템이나 다양한 클라우드 알람 서비스로 자동화 활용하기 위한 방법으로 간단하게 Python을 통해서 예시를 만들어 보고자 합니다.
Python을 사용하여 메일 발송 서버와 같이 데몬으로 25번 포트를 대기하고 메일 발송 요청이 수신되면 실제로 메일을 발송하지 않고 로그로 기록하는 코드를 작성할 수 있습니다. 이것은 간단한 SMTP 서버를 구현하는 것입니다. smtpd 모듈을 사용하여 이 작업을 수행할 수 있습니다. 먼저 smtpd 모듈을 가져와서 필요한 설정과 이벤트 핸들러를 구현해 보겠습니다.
import smtpd
import asyncore
import logging
# 로깅 설정
logging.basicConfig(filename="smtp_server.log", level=logging.INFO, format="%(asctime)s - %(message)s")
# SMTP 서버 설정
SMTP_SERVER_ADDRESS = ("localhost", 25) # 25번 포트 사용
SMTP_BANNER = "Custom SMTP Server"
class CustomSMTPServer(smtpd.SMTPServer):
def __init__(self, localaddr, remoteaddr):
super().__init__(localaddr, remoteaddr)
def process_message(self, peer, mailfrom, rcpttos, data):
logging.info(f"Received email from {peer}:")
logging.info(f"From: {mailfrom}")
logging.info(f"To: {rcpttos}")
logging.info(f"Message:\n{data}")
def run_server():
try:
server = CustomSMTPServer(SMTP_SERVER_ADDRESS, None)
logging.info(f"Starting SMTP server on {SMTP_SERVER_ADDRESS[0]}:{SMTP_SERVER_ADDRESS[1]}")
asyncore.loop()
except KeyboardInterrupt:
pass
if __name__ == "__main__":
run_server()이 코드는 SMTP 서버를 구현하고, 25번 포트에서 요청을 수신하면 이메일 데이터를 로그로 기록합니다. 이 코드를 실행하면 SMTP 서버가 실행되고, 이메일 발송 요청이 수신되면 로그 파일에 메시지가 기록됩니다. 주의할 점은 이 코드는 실제로 이메일을 발송하지 않고, 단순히 메일 내용을 로그에 기록합니다.
이 코드는 간단한 예제로서 보안 및 오류 처리를 포함하지 않습니다. 실제 환경에서 사용할 때에는 보안 및 예외 처리를 강화해야 합니다.
smtpd 모듈이 파이썬 3.12 버전에서 사용 중단되고 유지 관리되지 않으며, 파이썬 3.12에서는 제거될 예정임을 나타내는 경고 메시지입니다. 이 모듈은 파이썬에서 SMTP 서버를 구현하기 위한 라이브러리이며, 더 이상 사용되지 않으며 권장되지 않습니다.
이 에러를 해결하려면 대신 aiosmtpd 라이브러리를 사용하십시오. aiosmtpd는 asyncio를 사용하여 SMTP 서버를 구현하는 데 사용할 수 있는 라이브러리입니다. 아래는 aiosmtpd를 사용하여 SMTP 서버를 구현하는 예제 코드입니다.
먼저, aiosmtpd 라이브러리를 설치해야 합니다. 다음 명령을 사용하여 설치할 수 있습니다.
pip install aiosmtpd그런 다음, aiosmtpd를 사용하여 SMTP 서버를 구현할 수 있습니다.
import asyncio
from aiosmtpd.controller import Controller
# SMTP 서버 설정
SMTP_SERVER_ADDRESS = ("0.0.0.0", 25) # 25번 포트 사용
class CustomSMTPHandler:
async def handle_DATA(self, server, session, envelope):
peer = session.peer
mailfrom = envelope.mail_from
rcpttos = envelope.rcpt_tos
data = envelope.content.decode("utf-8")
# 이메일 데이터 로깅
with open("smtp_server.log", "a") as log_file:
log_file.write(f"Received email from {peer}:\n")
log_file.write(f"From: {mailfrom}\n")
log_file.write(f"To: {', '.join(rcpttos)}\n")
log_file.write(f"Message:\n{data}\n\n")
async def run_smtp_server():
controller = Controller(CustomSMTPHandler(), hostname=SMTP_SERVER_ADDRESS[0], port=SMTP_SERVER_ADDRESS[1])
controller.start()
print(f"SMTP server started on {SMTP_SERVER_ADDRESS[0]}:{SMTP_SERVER_ADDRESS[1]}")
try:
while True:
await asyncio.sleep(3600) # 이벤트 루프를 계속 유지하기 위한 임의의 대기 시간 설정
except KeyboardInterrupt:
controller.stop()
if __name__ == "__main__":
asyncio.run(run_smtp_server())이 코드는 aiosmtpd를 사용하여 SMTP 서버를 구현하고, 이메일 데이터를 로그로 기록합니다. aiosmtpd는 파이썬의 asyncio를 사용하므로 비동기 프로그래밍 모델을 따르는 것에 유의하세요.
위 코드에서는 try 블록 안에서 이벤트 루프를 무한히 유지하도록 하고, KeyboardInterrupt 예외가 발생하면 SMTP 서버를 중지하도록 설정했습니다. 이렇게 하면 이벤트 루프가 예외로 인해 종료되지 않고 계속 실행되며, 서버가 계속 실행됩니다. 필요에 따라 대기 시간을 조정하십시오.
SMTP 서버에 로컬로 telnet을 사용하여 메일을 발송하는 테스트를 수행하려면 다음 단계를 따르실 수 있습니다.
1. SMTP 서버를 실행한 Python 스크립트가 이미 실행 중이라고 가정합니다. 만약 실행 중이 아니라면, SMTP 서버를 실행합니다.
2. 터미널 또는 명령 프롬프트를 열고 다음 명령을 사용하여 로컬로 SMTP 서버에 telnet으로 연결합니다. 'localhost'는 SMTP 서버의 호스트 이름 또는 IP 주소로 변경할 수 있습니다.
telnet localhost 253. SMTP 서버에 연결되면 다음과 같은 화면이 표시됩니다.
Trying 127.0.0.1...
Connected to localhost.
Escape character is '^]'.
220 Custom SMTP Server4. 이제 SMTP 프로토콜에 따라 명령을 입력하여 이메일을 보낼 수 있습니다. SMTP 프로토콜은 다음과 같은 단계로 이뤄집니다.
- HELO/EHLO: SMTP 서버에 연결한 클라이언트가 자신을 소개합니다.
- MAIL FROM: 발신자 이메일 주소를 지정합니다.
- RCPT TO: 수신자 이메일 주소를 지정합니다.
- DATA: 이메일 내용을 입력합니다.
- . (마침표): 이메일 작성을 종료합니다.
다음은 간단한 예제입니다.
HELO example.com
MAIL FROM: <sender@example.com>
RCPT TO: <recipient@example.com>
DATA
Subject: Test Email
From: sender@example.com
To: recipient@example.com
Hello,
This is a test email sent via telnet.
.5. 이메일 내용을 모두 입력한 후에는 . (마침표)를 입력하고 엔터 키를 누르면 이메일이 SMTP 서버로 전송됩니다.
6. SMTP 서버가 이메일을 수락하면 다음과 같은 응답이 표시됩니다.
250 2.0.0 Ok: queued as ABC12345678이것으로 이메일 발송 테스트가 완료되었습니다. 이메일은 SMTP 서버에 성공적으로 전달되었습니다.
7. SMTP 서버의 로그 파일에는 해당 메일 내용이 기록되어 있습니다.
Received email from ('127.0.0.1', 47794):
From: sender@example.com
To: recipient@example.com
Message:
Subject: Test Email
From: sender@example.com
To: recipient@example.com
Hello,
This is a test email sent via telnet.참고: 이 테스트는 실제로 이메일을 보내지 않고 로컬로만 처리하며, 실제 이메일 서버와 다르게 인터넷을 통해 이메일을 전송하지 않습니다. SMTP 서버 구현에서 이메일 발송에 대한 실제 동작이 구현되지 않았음을 기억하세요.
위 부분은 네트워크 장비나 보안솔루션 등 메일로 발송된 내용을 메일시스템이 아닌 간단하게 파일로 기록하였지만, 현재 운영하는 방식에 따라 Elasticsearch이나 Slack 등 시스템으로 연동하여 통합 모니터링 관리 될 수 있도록 구현할 수 있습니다.
슬랙으로 전송하는 작업을 수행하려면 다음과 같은 단계가 필요합니다.
- SMTP 서버를 띄우고 메일을 수신하는 코드 작성
- 수신된 메일에서 수신자 메일 주소, 제목, 내용을 추출하는 코드 작성
- 추출된 정보를 사용하여 Slack으로 메시지를 보내는 코드 작성
아래는 이러한 작업을 수행하는 간단한 예제 코드입니다. 이 코드는 smtplib, email, slack_sdk 모듈을 사용합니다. 해당 모듈들은 미리 설치되어 있어야 합니다. 필요하다면 다음 명령어를 사용하여 설치할 수 있습니다.
pip install secure-smtplib
pip install slack_sdk그 후, 다음은 코드 예제입니다.
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from email import policy
from email.parser import BytesParser
from slack_sdk import WebClient
from slack_sdk.errors import SlackApiError
# SMTP 서버 설정
SMTP_SERVER = 'your_smtp_server'
SMTP_PORT = 587
SMTP_USERNAME = 'your_username'
SMTP_PASSWORD = 'your_password'
# Slack 설정
SLACK_API_TOKEN = 'your_slack_api_token'
SLACK_CHANNEL = '#your_channel'
# SMTP 서버 띄우기
def start_smtp_server():
with smtplib.SMTP(SMTP_SERVER, SMTP_PORT) as server:
server.login(SMTP_USERNAME, SMTP_PASSWORD)
server.starttls()
# 메일 수신 대기
server.listen()
print("SMTP Server listening...")
try:
while True:
connection, client_address = server.accept()
with connection:
print('Connection from', client_address)
# 메일 읽기
data = connection.recv(1024)
msg = BytesParser(policy=policy.default).parsebytes(data)
to_address = msg['to']
subject = msg['subject']
body = msg.get_body(preferencelist=('plain',)).get_content()
# Slack으로 메시지 보내기
send_to_slack(to_address, subject, body)
except KeyboardInterrupt:
print("SMTP Server stopped.")
# Slack으로 메시지 보내기
def send_to_slack(to_address, subject, body):
client = WebClient(token=SLACK_API_TOKEN)
try:
response = client.chat_postMessage(
channel=SLACK_CHANNEL,
text=f"New Email Received\nTo: {to_address}\nSubject: {subject}\nBody: {body}"
)
print("Slack message sent:", response['ts'])
except SlackApiError as e:
print(f"Error sending Slack message: {e.response['error']}")
if __name__ == "__main__":
start_smtp_server()이 코드는 SMTP 서버를 띄우고, 수신된 메일에서 수신자 메일 주소, 제목, 내용을 추출한 후 Slack으로 메시지를 보냅니다. 이 코드를 사용하기 전에 실제 서비스에 적합하도록 보안 및 예외 처리를 강화해야 합니다.
보안 시스템에서 발송되는 메일을 수신하고, 2단계 인증 코드를 포함한 다양한 메일을 효과적으로 자동화하여 처리하는 방안으로 IMAP(Internet Message Access Protocol) 또는 POP3(Post Office Protocol)를 사용하여 이메일을 수신하고, 이를 자동화하는 스크립트로, 특정 패턴의 메일(예: 2단계 인증 코드)을 처리하는 방법을 보여드리겠습니다.
1. IMAP을 통해 이메일 수신
Python의 imaplib 및 email 모듈을 사용하여 IMAP을 통해 이메일을 수신할 수 있습니다. 예제는 Gmail을 기준으로 작성되었습니다.
import imaplib
import email
from email.header import decode_header
# 이메일 계정 정보
username = "your-email@gmail.com"
password = "your-email-password"
# IMAP 서버 연결
mail = imaplib.IMAP4_SSL("imap.gmail.com")
mail.login(username, password)
# 받은 편지함 선택
mail.select("inbox")
# 모든 메일 검색
status, messages = mail.search(None, "ALL")
# 최신 메일 가져오기
mail_ids = messages[0].split()
latest_email_id = mail_ids[-1]
status, msg_data = mail.fetch(latest_email_id, "(RFC822)")
raw_email = msg_data[0][1]
msg = email.message_from_bytes(raw_email)
# 이메일 제목 및 발신자 파싱
subject, encoding = decode_header(msg["Subject"])[0]
if isinstance(subject, bytes):
subject = subject.decode(encoding if encoding else "utf-8")
from_ = msg.get("From")
print(f"Subject: {subject}")
print(f"From: {from_}")2. 이메일 필터링 및 파싱
이메일 내용을 파싱하여 특정 패턴을 찾아내고 이를 처리하는 예제입니다. 여기서는 2단계 인증 코드를 포함하는 이메일을 찾습니다.
import re
# 이메일 본문 파싱
def get_email_body(msg):
if msg.is_multipart():
for part in msg.walk():
if part.get_content_type() == "text/plain":
return part.get_payload(decode=True).decode()
else:
return msg.get_payload(decode=True).decode()
email_body = get_email_body(msg)
# 2단계 인증 코드 패턴 찾기 (예: 6자리 숫자)
pattern = r"\b\d{6}\b"
auth_code = re.search(pattern, email_body)
if auth_code:
print(f"2단계 인증 코드: {auth_code.group()}")
else:
print("2단계 인증 코드가 이메일에 포함되어 있지 않습니다.")3. 자동화된 작업 수행
이제 추출한 인증 코드를 사용하여 자동화된 작업을 수행할 수 있습니다. 예를 들어, 인증 코드를 웹사이트에 입력하여 자동 로그인을 시도할 수 있습니다.
import requests
def perform_2fa_login(auth_code):
url = "https://example.com/2fa-login"
data = {
"username": username,
"password": password,
"auth_code": auth_code
}
response = requests.post(url, data=data)
if response.status_code == 200:
print("2단계 인증 로그인 성공!")
else:
print("2단계 인증 로그인 실패!")
if auth_code:
perform_2fa_login(auth_code.group())전체 코드
위의 모든 단계를 하나의 스크립트로 결합하면 다음과 같습니다.
import imaplib
import email
from email.header import decode_header
import re
import requests
# 이메일 계정 정보
username = "your-email@gmail.com"
password = "your-email-password"
# IMAP 서버 연결
mail = imaplib.IMAP4_SSL("imap.gmail.com")
mail.login(username, password)
# 받은 편지함 선택
mail.select("inbox")
# 모든 메일 검색
status, messages = mail.search(None, "ALL")
# 최신 메일 가져오기
mail_ids = messages[0].split()
latest_email_id = mail_ids[-1]
status, msg_data = mail.fetch(latest_email_id, "(RFC822)")
raw_email = msg_data[0][1]
msg = email.message_from_bytes(raw_email)
# 이메일 제목 및 발신자 파싱
subject, encoding = decode_header(msg["Subject"])[0]
if isinstance(subject, bytes):
subject = subject.decode(encoding if encoding else "utf-8")
from_ = msg.get("From")
print(f"Subject: {subject}")
print(f"From: {from_}")
# 이메일 본문 파싱
def get_email_body(msg):
if msg.is_multipart():
for part in msg.walk():
if part.get_content_type() == "text/plain":
return part.get_payload(decode=True).decode()
else:
return msg.get_payload(decode=True).decode()
email_body = get_email_body(msg)
# 2단계 인증 코드 패턴 찾기 (예: 6자리 숫자)
pattern = r"\b\d{6}\b"
auth_code = re.search(pattern, email_body)
if auth_code:
print(f"2단계 인증 코드: {auth_code.group()}")
# 자동화된 작업 수행
def perform_2fa_login(auth_code):
url = "https://example.com/2fa-login"
data = {
"username": username,
"password": password,
"auth_code": auth_code
}
response = requests.post(url, data=data)
if response.status_code == 200:
print("2단계 인증 로그인 성공!")
else:
print("2단계 인증 로그인 실패!")
perform_2fa_login(auth_code.group())
else:
print("2단계 인증 코드가 이메일에 포함되어 있지 않습니다.")이 스크립트는 최신 이메일을 확인하고, 이메일 본문에서 2단계 인증 코드를 추출하며, 추출한 코드를 사용하여 자동으로 로그인 시도를 합니다. 이를 통해 2단계 인증 과정을 자동화할 수 있습니다.
댓글