Chat program
Intro
파이썬을 사용해 간단한 채팅 서버를 만들어 보고자 한다. 프로그램은 파이썬의 socket과 Thread를 사용했고, 멀티 스레드를 통해 실시간으로 정보를 수정하거나 여러 클라이언트의 요청을 수행한다.
주요 기능
Server
• 서버 실행 시 터미널에서 포트 번호를 입력받는다.
• registered_users.json 파일을 통해 기존 등록된 사용자 정보를 불러오며, 새로운 사용자는 최대 5명까지만 등록할 수 있다.
• 클라이언트가 접속하면 로그인 화면이 표시되며, /login <닉네임> 명령어로 원하는 닉네임을 입력할 수 있다.
• 비밀번호는 /pass <비밀번호> 명령어를 통해 설정(또는 로그인)할 수 있다.
• 로그인 후에는 메인 메뉴를 통해 닉네임 변경, 대화방 입장, 도움말 확인 등의 기능을 이용할 수 있다.
• 채팅방에서는 메시지 송수신 및 채팅 기록 관리가 이루어지며, 각 대화방의 기록은 최대 100줄까지 유지된다.Client
• 클라이언트 실행 시 접속할 서버의 IP 주소와 포트 번호를 입력받는다.
• 메시지 수신은 별도의 스레드에서 처리되며, ANSI 색상을 사용해 공지사항([NOTICE])과 채팅 메시지([CHAT])를 구분하여 출력한다.
• 사용자는 명령어를 통해 서버와 상호작용할 수 있다.Code
Server
import socket
import threading
import os
import json
HOST = '0.0.0.0'
port_input = input("서버 포트 번호를 입력하세요: ")
try:
PORT = int(port_input)
except ValueError:
print("유효한 포트 번호를 입력하세요.")
exit(1)
LOG_DIR = "chat_logs"
if not os.path.exists(LOG_DIR):
os.makedirs(LOG_DIR)
REGISTERED_USERS_FILE = "registered_users.json"
MAX_USERS = 5
# 파일에서 등록된 사용자 불러오기 (없으면 빈 dict)
if os.path.exists(REGISTERED_USERS_FILE):
try:
with open(REGISTERED_USERS_FILE, "r", encoding="utf-8") as f:
registered_users = json.load(f)
except Exception as e:
print("등록된 사용자 파일 로드 오류:", e)
registered_users = {}
else:
registered_users = {}
# key: client_socket, value: {"nickname": str, "state": "login"/"menu"/"chat", "room": str or None, "pending_nick": str}
clients = {}
lock = threading.Lock()
def save_registered_users():
try:
with open(REGISTERED_USERS_FILE, "w", encoding="utf-8") as f:
json.dump(registered_users, f, ensure_ascii=False, indent=2)
except Exception as e:
print("등록 사용자 저장 오류:", e)
def send(client, msg):
try:
client.sendall((msg + "\n").encode())
except Exception as e:
print("전송 에러:", e)
def save_chat_log(room, message):
log_file = os.path.join(LOG_DIR, f"{room}.txt")
try:
with open(log_file, "a+", encoding="utf-8") as f:
f.write(message + "\n")
f.seek(0)
lines = f.readlines()
if len(lines) > 100:
with open(log_file, "w", encoding="utf-8") as f:
f.writelines(lines[-100:])
except Exception as e:
print(f"채팅 기록 저장 오류({room}):", e)
def load_chat_log(room):
log_file = os.path.join(LOG_DIR, f"{room}.txt")
if os.path.exists(log_file):
try:
with open(log_file, "r", encoding="utf-8") as f:
return f.readlines()[-100:]
except Exception as e:
print(f"채팅 기록 로드 오류({room}):", e)
return []
def broadcast_to_room(room, message, exclude=None):
with lock:
recipients = [client for client, info in clients.items()
if info.get("state") == "chat" and info.get("room") == room and client != exclude]
for client in recipients:
send(client, message)
def send_main_menu(client):
with lock:
logged_in = [info["nickname"] for info in clients.values() if info.get("nickname")]
menu = (
"\n===== 메인 메뉴 =====\n"
"/nick <새닉네임> : 닉네임 변경\n"
"/join <대상유저> : 1:1 대화방 입장 (예: /join user3)\n"
" 또는 /join all (전체 톡방 입장)\n"
"/help : 사용 방법\n"
"/exit : 프로그램 종료\n"
f"\n현재 로그인된 유저: {', '.join(logged_in) if logged_in else '없음'}\n"
)
send(client, menu)
def broadcast_main_menu_update():
with lock:
menu_clients = [client for client, info in clients.items() if info.get("state") == "menu"]
for client in menu_clients:
send_main_menu(client)
def handle_client(client_socket, addr):
global registered_users
with lock:
clients[client_socket] = {"nickname": None, "state": "login", "room": None, "pending_nick": None}
send(client_socket, "=== 로그인 화면 ===")
send(client_socket, "로그인하려면: /login <닉네임> (예: /login myname)")
try:
while True:
msg = client_socket.recv(1024).decode().strip()
if not msg:
break
with lock:
state = clients[client_socket]["state"]
# ------ 로그인 상태 ------
if state == "login":
if msg.startswith("/login "):
desired = msg.split(" ", 1)[1].strip()
with lock:
if desired in registered_users:
clients[client_socket]["pending_nick"] = desired
response = f"{desired}는 기존 계정입니다. 비밀번호를 입력하세요: /pass <비밀번호>"
else:
if len(registered_users) >= MAX_USERS:
response = "등록 가능한 최대 유저 수에 도달했습니다. 로그인할 수 없습니다."
else:
clients[client_socket]["pending_nick"] = desired
response = f"{desired}는 새 계정입니다. 비밀번호를 설정해주세요: /pass <비밀번호>"
send(client_socket, response)
elif msg.startswith("/pass "):
with lock:
pending = clients[client_socket].get("pending_nick")
if not pending:
send(client_socket, "먼저 /login 명령어를 사용하세요.")
continue
password = msg.split(" ", 1)[1].strip()
with lock:
if pending not in registered_users:
registered_users[pending] = password
save_registered_users()
else:
if registered_users[pending] != password:
send(client_socket, "비밀번호가 틀렸습니다. 다시 /pass 명령어로 시도해주세요.")
continue
clients[client_socket]["nickname"] = pending
clients[client_socket]["state"] = "menu"
send(client_socket, "환영합니다!")
send(client_socket, "메인 메뉴로 이동합니다.")
send_main_menu(client_socket)
broadcast_main_menu_update()
else:
send(client_socket, "로그인 중에는 /login 또는 /pass 명령어만 사용 가능합니다.")
# ------ 메인 메뉴 상태 ------
elif state == "menu":
if msg.startswith("/nick "):
new_nick = msg.split(" ", 1)[1].strip()
with lock:
old = clients[client_socket]["nickname"]
if new_nick in registered_users:
send(client_socket, "이미 사용 중인 닉네임입니다.")
continue
registered_users[new_nick] = registered_users.pop(old)
save_registered_users()
clients[client_socket]["nickname"] = new_nick
send(client_socket, f"닉네임이 {new_nick}으로 변경되었습니다.")
broadcast_main_menu_update()
elif msg.startswith("/join "):
join_arg = msg.split(" ", 1)[1].strip()
with lock:
current_nick = clients[client_socket]["nickname"]
if join_arg.lower() == "all":
room = "all"
room_text = "전체 톡방"
else:
logged_in_users = [info["nickname"] for s, info in clients.items()
if info["nickname"] and info["nickname"] != current_nick]
if join_arg in logged_in_users:
room = "private:" + "-".join(sorted([current_nick, join_arg]))
room_text = f"{join_arg}와의 대화방"
else:
send(client_socket, "대화방 입장은 다른 로그인된 유저와의 1:1 채팅 또는 '/join all'만 지원합니다.")
continue
clients[client_socket]["state"] = "chat"
clients[client_socket]["room"] = room
history = load_chat_log(room)
for line in history:
send(client_socket, line.strip())
join_msg = f"[NOTICE]{current_nick}님이 {room_text}에 입장했습니다."
save_chat_log(room, join_msg)
broadcast_to_room(room, join_msg, exclude=client_socket)
elif msg == "/help":
help_text = (
"\n[도움말]\n"
"1. /nick <새닉네임> : 닉네임 변경\n"
"2. /join <대상유저> 또는 /join all : 대화방 입장\n"
"3. /quit 또는 /main : 대화방 나가기(메인화면으로 이동)\n"
"4. /exit : 프로그램 종료\n"
)
send(client_socket, help_text)
elif msg == "/exit":
send(client_socket, "프로그램을 종료합니다.")
break
else:
send(client_socket, "메인 메뉴에서는 /nick, /join, /help, /exit 명령어만 사용 가능합니다.")
# ------ 채팅(대화방) 상태 ------
elif state == "chat":
if msg in ("/quit", "/main"):
with lock:
room = clients[client_socket]["room"]
current_nick = clients[client_socket]["nickname"]
room_text = "전체 톡방" if room == "all" else "대화방"
quit_msg = f"[NOTICE]{current_nick}님이 {room_text}에서 나갔습니다."
broadcast_to_room(room, quit_msg, exclude=client_socket)
with lock:
clients[client_socket]["state"] = "menu"
clients[client_socket]["room"] = None
send_main_menu(client_socket)
else:
with lock:
room = clients[client_socket]["room"]
current_nick = clients[client_socket]["nickname"]
formatted_msg = f"[CHAT]{current_nick}: {msg}"
save_chat_log(room, formatted_msg)
broadcast_to_room(room, formatted_msg, exclude=client_socket)
except Exception as e:
print(f"클라이언트 처리 중 오류: {e}")
finally:
with lock:
client_info = clients.get(client_socket)
if client_info:
nick = client_info.get("nickname")
room = client_info.get("room")
if client_info.get("state") == "chat" and room:
room_text = "전체 톡방" if room == "all" else "대화방"
quit_msg = f"[NOTICE]{nick}님이 {room_text}에서 나갔습니다."
broadcast_to_room(room, quit_msg, exclude=client_socket)
del clients[client_socket]
client_socket.close()
def server_loop():
server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server_socket.bind((HOST, PORT))
server_socket.listen()
print(f"서버가 {PORT} 포트에서 대기 중...")
try:
while True:
client_sock, addr = server_socket.accept()
threading.Thread(target=handle_client, args=(client_sock, addr), daemon=True).start()
except KeyboardInterrupt:
print("서버 종료 신호 받음. 모든 클라이언트에 종료 메시지를 전송합니다.")
shutdown_msg = "[SERVER SHUTDOWN] 서버가 종료되었습니다. 프로그램을 종료합니다."
with lock:
clients_list = list(clients.keys())
for client in clients_list:
try:
send(client, shutdown_msg)
client.close()
except Exception as e:
print("클라이언트 종료 에러:", e)
server_socket.close()
if __name__ == "__main__":
server_loop()Client
import socket
import threading
import sys
# ANSI escape sequences
BLUE = "\033[34m"
RED = "\033[31m"
RESET = "\033[0m"
def receive_messages(client_socket):
while True:
try:
msg = client_socket.recv(1024).decode()
if msg:
if msg.startswith("[SERVER SHUTDOWN]"):
print(RED + msg[18:] + RESET)
sys.exit(0)
elif msg.startswith("[NOTICE]"):
print(RED + msg[8:] + RESET)
elif msg.startswith("[CHAT]"):
print(BLUE + msg[6:] + RESET)
else:
print(msg)
else:
break
except Exception:
break
def main():
host = input("서버 IP 주소를 입력하세요 (기본: 127.0.0.1): ") or "127.0.0.1"
port_input = input("서버 포트 번호를 입력하세요: ")
try:
port = int(port_input)
except ValueError:
print("유효한 포트 번호를 입력하세요.")
return
client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
try:
client_socket.connect((host, port))
except Exception as e:
print("서버 연결 실패:", e)
return
recv_thread = threading.Thread(target=receive_messages, args=(client_socket,), daemon=True)
recv_thread.start()
while True:
try:
user_input = input()
if user_input:
client_socket.sendall(user_input.encode())
if user_input == "/exit":
break
except KeyboardInterrupt:
break
client_socket.close()
sys.exit(0)
if __name__ == "__main__":
main()
보안 공부 로그