API安全防禦深度解析與威脅防範

import bleach

def sanitize_html(input_html):
    allowed_tags = ['b', 'i', 'u', 'a', 'p']
    allowed_attributes = {'a': ['href', 'title']}
    return bleach.clean(input_html, tags=allowed_tags, attributes=allowed_attributes)

# 使用範例：
safe_output = sanitize_html(user_provided_html)

import os
import hmac
import hashlib
from flask import Flask, request, jsonify, make_response

app = Flask(__name__)
app.config['CSRF_SECRET'] = os.urandom(32)

def generate_csrf_token():
    return hmac.new(app.config['CSRF_SECRET'], os.urandom(16), hashlib.sha256).hexdigest()

@app.before_request
def verify_csrf_token():
    if request.method in ['POST', 'PUT', 'DELETE']:
        csrf_cookie = request.cookies.get('csrf_token')
        csrf_header = request.headers.get('X-CSRF-Token')
        if not csrf_cookie or not csrf_header or not hmac.compare_digest(csrf_cookie, csrf_header):
            return make_response(jsonify({"error": "CSRF token validation failed"}), 403)

@app.route('/get-token', methods=['GET'])
def get_token():
    token = generate_csrf_token()
    resp = make_response(jsonify({"csrf_token": token}))
    resp.set_cookie('csrf_token', token, httponly=True, secure=True)
    return resp

# 使用範例：
@app.route('/update', methods=['POST'])
def update_resource():
    return jsonify({"status": "updated"})

if __name__ == '__main__':
    app.run()

from sqlalchemy import create_engine, text

engine = create_engine('postgresql://user:password@localhost/dbname', echo=False)

def fetch_user_details(user_identifier):
    sql_query = text("SELECT * FROM users WHERE id = :user_id")
    result = engine.execute(sql_query, user_id=user_identifier)
    return result.fetchone()

import logging
import json

# 組態結構化日誌記錄
logger = logging.getLogger('secure_api')
handler = logging.StreamHandler()
formatter = logging.Formatter('%(asctime)s %(name)s %(levelname)s %(message)s')
handler.setFormatter(formatter)
logger.addHandler(handler)
logger.setLevel(logging.INFO)

def log_security_event(event_type, detail):
    logger.info(json.dumps({"event_type": event_type, "detail": detail}))

import time
import redis

r = redis.StrictRedis(host='localhost', port=6379, db=0)

def token_bucket(client_id, rate, capacity):
    """
    為每個客戶端實施令牌桶演算法。
    :param client_id: 客戶端的唯一識別符。
    :param rate: 每秒新增的令牌數量。
    :param capacity: 令牌桶的最大容量。
    :return: 若請求被允許則傳回True，否則傳回False。
    """
    key = f"rate_limit:{client_id}"
    now = time.time()
    pipeline = r.pipeline()
    
    # 檢索目前的令牌數量和上次補充時間
    pipeline.hmget(key, "tokens", "last_refill")
    tokens, last_refill = pipeline.execute()[0]
    if tokens is None or last_refill is None:
        # 若未初始化，則初始化令牌桶
        tokens = capacity
        last_refill = now
    else:
        tokens = float(tokens)
        last_refill = float(last_refill)
    
    # 根據經過的時間補充令牌
    elapsed = now - last_refill
    tokens = min(capacity, tokens + elapsed * rate)
    if tokens < 1:
        # 若令牌不足，則拒絕請求
        return False
    else:
        tokens -= 1
    
    # 更新Redis中的狀態
    pipeline.hmset(key, {"tokens": tokens, "last_refill": now})
    pipeline.expire(key, int(capacity / rate * 2))
    pipeline.execute()
    return True

# 使用範例：
# if token_bucket("client123", rate=5, capacity=10):
#     process_request()
# else:
#     return error_response("超出速率限制")

import time
import redis

r = redis.StrictRedis(host='localhost', port=6379, db=0)

def fixed_window_rate_limit(client_id, limit, window_size):
    """
    實施固定視窗速率限制演算法。
    :param client_id: 客戶端的唯一識別符。
    :param limit: 每個視窗允許的最大請求數。
    :param window_size: 時間視窗的大小（秒）。
    :return: 若請求被接受則傳回True，否則傳回False。
    """
    key = f"rate_limit:{client_id}:{int(time.time() // window_size)}"
    count = r.get(key)
    if count is None:
        r.set(key, 1, ex=window_size)
        return True
    elif int(count) < limit:
        r.incr(key)
        return True
    else:
        return False

# 使用範例：
# if fixed_window_rate_limit("client123", limit=10, window_size=60):
#     process_request()
# else:
#     return error_response("超出速率限制")