파이썬을 사용하여 간단한 보안 모듈 만들기 250106

q1: 파이썬을 사용하여 간단한 보안 모듈 만들기

a1:

파이썬을 사용하여 간단한 보안 모듈을 만들기 위해 사용자 인증, 권한 관리, 정책 엔진, 로깅 및 모니터링 기능을 구현하는 예제를 상세히 설명하겠습니다. 이 보안 모듈은 사용자 요청을 확인하고 권한을 검증한 뒤, 데이터베이스에 요청을 전달하거나 차단하는 역할을 합니다.

1. 기능 정의
1. 사용자 인증:
• JWT(Json Web Token)를 사용하여 사용자 인증.
2. 권한 관리:
• 역할(Role) 기반 접근 제어(RBAC).
• 사용자는 역할(Role)을 가지고, 역할은 권한(Policy)을 정의.
3. 정책 엔진:
• 사용자가 요청한 리소스와 권한을 비교해 허용/차단.
4. 로깅:
• 모든 요청 및 정책 위반을 로그로 기록.

2. 기술 스택
• Python Framework: Flask
• JWT Library: PyJWT
• Database: SQLite (간단한 데이터 저장용)
• Logging: Python 내장 logging 모듈

3. 프로젝트 구조

security_module/
├── app.py                 # 메인 애플리케이션
├── auth.py                # 인증 관련 모듈
├── policy_engine.py       # 정책 엔진 모듈
├── db.py                  # 데이터베이스 초기화 및 쿼리
├── logs/                  # 요청 및 정책 위반 로그 저장 폴더
│   └── security.log
└── requirements.txt       # 필요 라이브러리

4. 구현

(1) 데이터베이스 초기화

db.py: 사용자, 역할, 정책 정보를 저장합니다.

import sqlite3

def init_db():
    conn = sqlite3.connect('security.db')
    cursor = conn.cursor()

    # 사용자 테이블
    cursor.execute('''
    CREATE TABLE IF NOT EXISTS users (
        id INTEGER PRIMARY KEY,
        username TEXT NOT NULL UNIQUE,
        password TEXT NOT NULL,
        role TEXT NOT NULL
    )
    ''')

    # 정책 테이블
    cursor.execute('''
    CREATE TABLE IF NOT EXISTS policies (
        id INTEGER PRIMARY KEY,
        role TEXT NOT NULL,
        resource TEXT NOT NULL,
        action TEXT NOT NULL
    )
    ''')

    # 초기 데이터 입력
    cursor.execute("INSERT OR IGNORE INTO users (username, password, role) VALUES ('admin', 'admin123', 'admin')")
    cursor.execute("INSERT OR IGNORE INTO policies (role, resource, action) VALUES ('admin', 'database', 'read')")
    cursor.execute("INSERT OR IGNORE INTO policies (role, resource, action) VALUES ('admin', 'database', 'write')")

    conn.commit()
    conn.close()

def query_db(query, params=()):
    conn = sqlite3.connect('security.db')
    cursor = conn.cursor()
    cursor.execute(query, params)
    results = cursor.fetchall()
    conn.commit()
    conn.close()
    return results

(2) 사용자 인증

auth.py: JWT 기반 인증 구현.

import jwt
import datetime
from flask import request, jsonify

SECRET_KEY = "secret_key"

def generate_token(username):
    payload = {
        "username": username,
        "exp": datetime.datetime.utcnow() + datetime.timedelta(hours=1)  # 토큰 만료 시간
    }
    return jwt.encode(payload, SECRET_KEY, algorithm="HS256")

def decode_token(token):
    try:
        return jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
    except jwt.ExpiredSignatureError:
        return None

def authenticate():
    auth_header = request.headers.get("Authorization")
    if not auth_header:
        return None
    token = auth_header.split(" ")[1]
    return decode_token(token)

(3) 정책 엔진

policy_engine.py: 사용자 요청을 검증하는 모듈.

from db import query_db

def is_allowed(role, resource, action):
    """
    역할, 리소스, 작업을 기반으로 권한 확인.
    """
    results = query_db(
        "SELECT * FROM policies WHERE role = ? AND resource = ? AND action = ?",
        (role, resource, action)
    )
    return len(results) > 0

(4) 메인 애플리케이션

app.py: Flask 애플리케이션 설정.

from flask import Flask, request, jsonify
from auth.py import generate_token, authenticate
from policy_engine import is_allowed
from db import init_db, query_db
import logging

# Flask 앱 초기화
app = Flask(__name__)

# 로그 설정
logging.basicConfig(filename='logs/security.log', level=logging.INFO)

# 데이터베이스 초기화
init_db()

@app.route('/login', methods=['POST'])
def login():
    data = request.json
    username = data.get('username')
    password = data.get('password')

    user = query_db("SELECT * FROM users WHERE username = ? AND password = ?", (username, password))
    if user:
        token = generate_token(username)
        return jsonify({"token": token})
    return jsonify({"message": "Invalid credentials"}), 401

@app.route('/resource/<resource>', methods=['POST'])
def access_resource(resource):
    token_data = authenticate()
    if not token_data:
        return jsonify({"message": "Unauthorized"}), 401

    username = token_data["username"]
    user = query_db("SELECT * FROM users WHERE username = ?", (username,))
    role = user[0][3]  # 역할 정보 가져오기

    action = request.json.get('action')

    if is_allowed(role, resource, action):
        logging.info(f"Access granted: {username} -> {resource} ({action})")
        return jsonify({"message": "Access granted"})
    else:
        logging.warning(f"Access denied: {username} -> {resource} ({action})")
        return jsonify({"message": "Access denied"}), 403

if __name__ == "__main__":
    app.run(debug=True)

(5) 요구 사항 설치

requirements.txt:

Flask
PyJWT

설치 명령:

pip install -r requirements.txt

5. 실행 및 테스트
1. 서버 실행:

python app.py


2. 로그인 요청:

curl -X POST http://127.0.0.1:5000/login -H "Content-Type: application/json" -d '{"username": "admin", "password": "admin123"}'


3. 자원 접근 요청:

curl -X POST http://127.0.0.1:5000/resource/database \
     -H "Authorization: Bearer <your_token>" \
     -H "Content-Type: application/json" \
     -d '{"action": "read"}'

6. 개선 및 확장
• 다중 사용자 및 역할 지원: 역할 및 정책 테이블을 확장.
• 로깅 강화: 로그 데이터를 Elasticsearch와 같은 외부 시스템에 통합.
• 데이터 암호화: 사용자 정보 암호화 저장.

이 보안 모듈은 간단하지만, 실제 보안 시스템의 기본적인 흐름을 이해하고 확장할 수 있는 기반을 제공합니다.