Python黑白名单案例如何配置访问黑白名单

wen python案例 27

本文目录导读:

Python黑白名单案例如何配置访问黑白名单

  1. 基于IP的黑白名单
  2. 基于用户/角色的黑白名单
  3. 基于API路径的黑白名单
  4. 配置文件方式(YAML/JSON)
  5. 综合实现(带缓存和动态更新)
  6. 最佳实践建议

我来详细讲解Python中配置黑白名单的几种常见案例和实现方式。

基于IP的黑白名单

简单列表实现

class IPAccessControl:
    def __init__(self):
        # 白名单
        self.white_list = {
            '192.168.1.100',
            '192.168.1.101',
            '10.0.0.1'
        }
        # 黑名单
        self.black_list = {
            '192.168.1.200',
            '10.0.0.100',
            '172.16.0.50'
        }
        # 是否启用白名单模式
        self.white_list_mode = False
    def check_access(self, ip_address):
        """检查IP是否允许访问"""
        # 1. 先检查黑名单
        if ip_address in self.black_list:
            return False, "IP地址在黑名单中"
        # 2. 如果启用白名单模式
        if self.white_list_mode:
            if ip_address in self.white_list:
                return True, "IP地址在白名单中"
            else:
                return False, "IP地址不在白名单中"
        return True, "IP地址允许访问"
    def add_to_white_list(self, ip_address):
        """添加IP到白名单"""
        self.white_list.add(ip_address)
        # 如果已存在于黑名单,移除
        self.black_list.discard(ip_address)
    def add_to_black_list(self, ip_address):
        """添加IP到黑名单"""
        self.black_list.add(ip_address)
        # 如果已存在于白名单,移除
        self.white_list.discard(ip_address)
# 使用示例
access_control = IPAccessControl()
# 测试不同的IP
test_ips = [
    '192.168.1.100',  # 白名单
    '192.168.1.200',  # 黑名单
    '192.168.1.50',   # 其他IP
]
for ip in test_ips:
    allowed, message = access_control.check_access(ip)
    print(f"IP: {ip} - 允许访问: {allowed} - 原因: {message}")
# 启用白名单模式
access_control.white_list_mode = True
print("\n--- 启用白名单模式 ---")
for ip in test_ips:
    allowed, message = access_control.check_access(ip)
    print(f"IP: {ip} - 允许访问: {allowed} - 原因: {message}")

使用CIDR支持子网

import ipaddress
class IPAccessControlCIDR:
    def __init__(self):
        self.white_list = [
            ipaddress.ip_network('192.168.1.0/24'),
            ipaddress.ip_network('10.0.0.0/8'),
        ]
        self.black_list = [
            ipaddress.ip_network('192.168.1.200/32'),
            ipaddress.ip_network('10.10.0.0/16'),
        ]
    def check_access(self, ip_address):
        ip = ipaddress.ip_address(ip_address)
        # 检查黑名单
        for network in self.black_list:
            if ip in network:
                return False, "IP地址在黑名单中"
        # 检查白名单
        for network in self.white_list:
            if ip in network:
                return True, "IP地址在白名单中"
        return False, "IP地址不在白名单中"
# 使用示例
cidr_control = IPAccessControlCIDR()
test_ip = '192.168.1.50'
allowed, message = cidr_control.check_access(test_ip)
print(f"IP: {test_ip} - {message}")

基于用户/角色的黑白名单

class UserAccessControl:
    def __init__(self):
        self.user_white_list = {
            'admin',
            'superuser',
            'manager'
        }
        self.user_black_list = {
            'banned_user1',
            'banned_user2'
        }
        # 角色权限配置
        self.role_permissions = {
            'admin': ['read', 'write', 'delete', 'manage'],
            'user': ['read', 'write'],
            'guest': ['read'],
            'banned': []  # 被禁止的角色
        }
    def check_user_access(self, username, required_permission=None):
        """检查用户访问权限"""
        # 检查黑名单
        if username in self.user_black_list:
            return False, "用户已被禁止访问"
        # 检查白名单(可选)
        if username in self.user_white_list:
            return True, "用户在白名单中,拥有所有权限"
        # 检查角色权限
        user_role = self.get_user_role(username)
        if user_role == 'banned':
            return False, "用户角色已被禁止"
        if required_permission:
            if required_permission in self.role_permissions.get(user_role, []):
                return True, f"用户有{required_permission}权限"
            else:
                return False, f"用户没有{required_permission}权限"
        return True, "用户允许访问"
    def get_user_role(self, username):
        """模拟获取用户角色"""
        # 实际应用中这里会查询数据库
        roles = {
            'alice': 'admin',
            'bob': 'user',
            'charlie': 'guest',
            'mallory': 'banned'
        }
        return roles.get(username, 'guest')
# 使用示例
user_control = UserAccessControl()
users = ['alice', 'bob', 'charlie', 'mallory', 'unknown']
for user in users:
    allowed, message = user_control.check_user_access(user, 'read')
    print(f"用户: {user} - 允许访问: {allowed} - {message}")

基于API路径的黑白名单

from flask import Flask, request, jsonify
from functools import wraps
import re
app = Flask(__name__)
class APIAccessControl:
    def __init__(self):
        # API白名单(允许访问的路径模式)
        self.api_white_list = [
            r'^/api/public/.*',
            r'^/api/health$',
            r'^/api/version$',
        ]
        # API黑名单(禁止访问的路径模式)
        self.api_black_list = [
            r'^/api/admin/.*',
            r'^/api/internal/.*',
            r'^/api/debug/.*',
        ]
        # 特定IP对特定API的访问控制
        self.ip_api_rules = {
            '192.168.1.0/24': {
                'white_list': [r'^/api/public/.*'],
                'black_list': []
            }
        }
    def check_api_access(self, path, ip_address):
        """检查API访问权限"""
        # 检查黑名单
        for pattern in self.api_black_list:
            if re.match(pattern, path):
                return False, "API路径在黑名单中"
        # 如果启用白名单模式,检查白名单
        if self.api_white_list:
            for pattern in self.api_white_list:
                if re.match(pattern, path):
                    return True, "API路径允许访问"
            return False, "API路径不在白名单中"
        return True, "API路径允许访问"
# Flask装饰器实现访问控制
def require_whitelist():
    def decorator(f):
        @wraps(f)
        def decorated_function(*args, **kwargs):
            access_control = APIAccessControl()
            ip = request.remote_addr
            path = request.path
            allowed, message = access_control.check_api_access(path, ip)
            if not allowed:
                return jsonify({
                    'error': 'access_denied',
                    'message': message
                }), 403
            return f(*args, **kwargs)
        return decorated_function
    return decorator
@app.route('/api/public/data')
@require_whitelist()
def public_data():
    return jsonify({'data': '公开数据'})
@app.route('/api/admin/config')
@require_whitelist()
def admin_config():
    return jsonify({'config': '管理员配置'})

配置文件方式(YAML/JSON)

white_black_list.yaml

# 黑白名单配置
access_control:
  # 模式: 'strict' (严格模式,只允许白名单) 或 'loose' (宽松模式,只阻止黑名单)
  mode: strict
  # IP黑白名单
  ip:
    white_list:
      - 192.168.1.0/24
      - 10.0.0.1/32
      - 172.16.0.0/16
    black_list:
      - 192.168.1.200/32
      - 10.0.0.100/32
  # 用户黑白名单
  user:
    white_list:
      - admin
      - superuser
    black_list:
      - banned_user
  # API路径黑白名单
  api:
    white_list:
      - /api/v1/public/*
      - /api/v1/health
    black_list:
      - /api/v1/admin/*
      - /api/v1/internal/*

加载配置文件

import yaml
import ipaddress
import fnmatch
class ConfigBasedAccessControl:
    def __init__(self, config_file='white_black_list.yaml'):
        with open(config_file, 'r') as f:
            self.config = yaml.safe_load(f)
        self.mode = self.config['access_control']['mode']
        self._parse_ip_lists()
        self._parse_user_lists()
        self._parse_api_lists()
    def _parse_ip_lists(self):
        """解析IP列表"""
        ip_config = self.config['access_control']['ip']
        self.ip_white_list = [
            ipaddress.ip_network(ip) 
            for ip in ip_config['white_list']
        ]
        self.ip_black_list = [
            ipaddress.ip_network(ip) 
            for ip in ip_config['black_list']
        ]
    def _parse_user_lists(self):
        """解析用户列表"""
        user_config = self.config['access_control']['user']
        self.user_white_list = set(user_config['white_list'])
        self.user_black_list = set(user_config['black_list'])
    def _parse_api_lists(self):
        """解析API列表"""
        api_config = self.config['access_control']['api']
        self.api_white_list = api_config['white_list']
        self.api_black_list = api_config['black_list']
    def check_ip(self, ip_address):
        """检查IP"""
        ip = ipaddress.ip_address(ip_address)
        # 检查黑名单
        for network in self.ip_black_list:
            if ip in network:
                return False
        # 严格模式:检查是否在白名单中
        if self.mode == 'strict':
            for network in self.ip_white_list:
                if ip in network:
                    return True
            return False
        return True
    def check_api(self, path):
        """检查API路径"""
        # 检查黑名单
        for pattern in self.api_black_list:
            if fnmatch.fnmatch(path, pattern):
                return False
        # 严格模式
        if self.mode == 'strict':
            for pattern in self.api_white_list:
                if fnmatch.fnmatch(path, pattern):
                    return True
            return False
        return True
# 使用示例
config_control = ConfigBasedAccessControl()
print(config_control.check_ip('192.168.1.50'))  # True
print(config_control.check_api('/api/v1/admin/config'))  # False

综合实现(带缓存和动态更新)

import time
import threading
from functools import lru_cache
class AdvancedAccessControl:
    def __init__(self):
        self.white_list = set()
        self.black_list = set()
        self.white_list_mode = False
        # 缓存配置
        self.cache_ttl = 300  # 5分钟
        self.cache = {}
        # 动态更新锁
        self.lock = threading.RLock()
        # 更新线程
        self.update_thread = threading.Thread(target=self._update_lists_periodically)
        self.update_thread.daemon = True
        self.update_thread.start()
    @lru_cache(maxsize=1000)
    def _check_cached(self, key):
        """带缓存的检查"""
        return self._check_internal(key)
    def _check_internal(self, key):
        """内部检查逻辑"""
        with self.lock:
            if key in self.black_list:
                return False
            if self.white_list_mode and key not in self.white_list:
                return False
            return True
    def check_access(self, key):
        """检查访问权限(带缓存)"""
        return self._check_cached(key)
    def add_to_white_list(self, key):
        """动态添加白名单"""
        with self.lock:
            self.white_list.add(key)
            self._check_cached.cache_clear()
    def remove_from_black_list(self, key):
        """动态移除黑名单"""
        with self.lock:
            self.black_list.discard(key)
            self._check_cached.cache_clear()
    def _update_lists_periodically(self):
        """定期更新列表(从数据库或配置文件)"""
        while True:
            time.sleep(60)  # 每分钟更新一次
            self._load_lists_from_database()
            self._check_cached.cache_clear()
    def _load_lists_from_database(self):
        """从数据库加载黑白名单"""
        # 实际应用中这里会连接数据库
        pass
# 使用示例
advanced_control = AdvancedAccessControl()
advanced_control.add_to_white_list('192.168.1.100')
print(advanced_control.check_access('192.168.1.100'))  # True

最佳实践建议

  1. 使用缓存:频繁访问的检查应该使用缓存
  2. 支持子网:不仅是单个IP,还要支持CIDR
  3. 动态更新:支持运行时更新黑白名单
  4. 日志记录:记录所有访问尝试和被拒绝的请求
  5. 性能优化:使用集合、哈希表等高效数据结构
  6. 分级控制:不同级别的资源使用不同的黑白名单
  7. 默认拒绝:安全最佳实践是默认拒绝,只允许白名单

选择哪种方式取决于你的具体需求,建议从简单的开始,根据需要逐渐增加复杂性。

抱歉,评论功能暂时关闭!