Django ABAC动态策略引擎实现零信任访问控制-北京尧图网络

简介本资源是一套基于Django框架实现的零信任网络访问控制系统完整源码面向计算机、通信、人工智能及自动化等专业的学生与教师适用于毕业设计、课程大作业或课程设计场景尤其适合具备Python基础并希望深入理解零信任架构落地实践的学习者。压缩包共33个文件含23个Python核心模块涵盖models、views、routing、middleware等Django关键组件、6个XML配置文件用于IDE或项目元数据管理、1个SQLite3数据库文件已预置初始数据、1个.gitignore及1个HTML模板页整体仅17KB轻量易部署。已有154人学习下载资源经作者调试验证答辩评分达95分代码结构清晰、模块职责分明sdpmanager_backend为服务端主应用log_manager负责审计日志consumers.py支持WebSocket实时策略下发templates/test.html提供简易前端交互界面配套utils与customEncoder.py体现序列化与安全编码细节具备扎实的工程参考价值与二次开发基础。1. 这不是另一个 Django 权限中间件它用动态策略引擎把“谁能在何时访问什么”拆解成可验证、可审计、可回滚的原子操作零信任在高校课程设计里常被简化为“登录后才能看页面”但这个毕设项目真正落地了 NIST SP 800-207 定义的三大支柱——持续验证、最小权限、基于属性的访问控制ABAC。它不依赖静态角色如is_staffTrue而是把每次 HTTP 请求拆解为「主体用户设备指纹地理位置→ 行为GET/POST/DELETE→ 资源API 路径数据标签敏感等级→ 环境时间窗口网络段TLS 版本」四元组交由PolicyEngine实时决策。答辩 95 分的关键在于所有策略变更都生成不可篡改的审计日志含决策链快照且支持策略热加载——改完 YAML 配置无需重启服务。适合两类人一是想透彻理解 ABAC 在 Web 框架中如何与 Django ORM、中间件、信号机制深度耦合的学生二是需要快速验证零信任落地路径的中小团队开发者——它没用任何商业 SDK纯 Python 实现连 SQLite 的db.sqlite3都预置了带标签的测试数据。2. 从策略建模到请求拦截Django 中间件层如何实现细粒度 ABAC 决策链零信任的核心不是“禁止未授权访问”而是“证明每一次访问都符合当前策略”。这个项目把抽象原则转化为可执行的代码结构关键在于三层解耦策略定义层YAML、策略执行层Python 类、策略注入层Django 中间件。下面拆解其技术选型逻辑与可复现步骤。2.1 策略定义用 YAML 描述动态上下文而非硬编码 if-else项目未使用 Django 内置的Permission模型而是将策略存于sdpmanager_backend/policies/目录下的 YAML 文件如api_access_policy.yaml。这种设计规避了数据库 Schema 变更带来的迁移风险且便于版本控制。一个典型策略如下# sdpmanager_backend/policies/api_access_policy.yaml policy_id: api_read_sensitive_data description: 仅允许认证用户在工作时间9-18点通过公司内网IP访问标记为sensitive的数据 subjects: - type: user attributes: - is_authenticated: true - groups__name: data_analyst - type: device attributes: - os_version: 10.15 - has_mfa_enabled: true resources: - type: api_endpoint path: /api/v1/users/ method: GET tags: [sensitive] conditions: - type: time_window start: 09:00 end: 18:00 - type: ip_range cidr: 10.0.0.0/8 effect: allow提示YAML 中的tags: [sensitive]对应models.py中DataResource模型的tags字段该字段是 JSONField支持多标签组合查询。策略解析器会将tags转为Q(tags__contains[sensitive])查询条件避免全表扫描。2.2 策略执行PolicyEngine 如何将 YAML 映射为可执行的 Python 对象策略加载由sdpmanager_backend/utils/policy_engine.py中的PolicyEngine类完成。它不是简单地yaml.load()而是构建了策略缓存树Trie按method path前缀索引加速匹配。核心逻辑如下# sdpmanager_backend/utils/policy_engine.py class PolicyEngine: def __init__(self): self._cache {} # {f{method}:{path_prefix}: [PolicyObject, ...]} self._load_policies() def _load_policies(self): for policy_file in Path(sdpmanager_backend/policies).glob(*.yaml): with open(policy_file) as f: raw_policies yaml.safe_load(f) for raw in raw_policies.get(policies, []): # 将 YAML 中的 subjects/resources/conditions 转为 callable policy PolicyObject( policy_idraw[policy_id], subjectsself._build_subject_evaluators(raw[subjects]), resourcesself._build_resource_evaluators(raw[resources]), conditionsself._build_condition_evaluators(raw[conditions]), effectraw[effect] ) # 按 method:path_prefix 缓存例如 GET:/api/v1/ cache_key f{raw[resources][0][method]}:{raw[resources][0][path].split(/)[1]} self._cache.setdefault(cache_key, []).append(policy) def evaluate(self, request: HttpRequest, resource_tags: List[str]) - bool: cache_key f{request.method}:{request.path.split(/)[1]} for policy in self._cache.get(cache_key, []): if (self._evaluate_subjects(policy.subjects, request) and self._evaluate_resources(policy.resources, request, resource_tags) and self._evaluate_conditions(policy.conditions, request)): return policy.effect allow return False # 默认拒绝2.2.1 主体评估器Subject Evaluator的实现细节_build_subject_evaluators()将subjects列表中的字符串如groups__name: data_analyst编译为闭包函数。关键点在于它不直接调用user.groups.filter(namedata_analyst)而是构造Q对象最终合并为单次数据库查询def _build_subject_evaluator(self, subject_str: str) - Callable[[HttpRequest], bool]: # 解析 groups__name: data_analyst → field_pathgroups__name, valuedata_analyst field_path, value self._parse_subject_rule(subject_str) # 构造 Q 对象避免 N1 查询 q_obj Q(**{f{field_path}: value}) return lambda req: req.user.is_authenticated and req.user.groups.filter(q_obj).exists()注意req.user.groups.filter(...).exists()是一次 SQL 查询比req.user.groups.all()后 Python 循环判断高效 10 倍以上。项目在views.py中所有需鉴权的视图都显式调用PolicyEngine().evaluate(request, resource_tags)而非依赖装饰器——这保证了资源标签如sensitive能动态传入而非写死在 URL 路由中。2.3 策略注入自定义中间件如何劫持请求并注入决策结果策略执行必须在 Django 请求生命周期的早期介入否则可能绕过鉴权如静态文件或csrf_exempt视图。项目在sdpmanager_backend/middleware.py中实现了ABACMiddleware其process_view方法在视图函数执行前触发# sdpmanager_backend/middleware.py class ABACMiddleware: def __init__(self, get_response): self.get_response get_response self.policy_engine PolicyEngine() # 单例避免重复加载 def process_view(self, request, view_func, view_args, view_kwargs): # 1. 提取当前请求对应的资源标签从视图函数的 docstring 或 model meta 获取 resource_tags self._extract_resource_tags(view_func) # 2. 执行 ABAC 决策 is_allowed self.policy_engine.evaluate(request, resource_tags) # 3. 将决策结果注入 request供视图内部使用如返回不同字段 request.abac_decision { allowed: is_allowed, reason: policy_denied if not is_allowed else policy_granted, matched_policy: self.policy_engine.last_matched_policy_id } # 4. 若拒绝立即返回 403不执行视图 if not is_allowed: return HttpResponseForbidden(Access denied by zero-trust policy) return None # 继续执行视图2.3.1 资源标签提取的两种可靠方式_extract_resource_tags()方法采用双重 fallback 策略确保标签来源可追溯优先级 1视图函数的__doc__字符串在views.py中每个视图函数顶部添加注释如def user_list_view(request): Tags: sensitive, pii Returns list of users with PII fields. # ...优先级 2模型 Meta 的abac_tags属性在models.py中定义class User(models.Model): name models.CharField(max_length100) email models.EmailField() class Meta: abac_tags [sensitive, pii] # 自定义 Meta 属性提示ABACMiddleware在settings.py的MIDDLEWARE列表中必须置于AuthenticationMiddleware之后、CommonMiddleware之前否则request.user未初始化或request.path被重写。项目settings.py第 62 行已正确配置sdpmanager_backend.middleware.ABACMiddleware,。3. 数据层与策略联动如何让 SQLite 中的资源标签驱动实时访问控制零信任的“持续验证”要求策略能感知资源本身的敏感属性而非仅依赖用户角色。该项目将资源敏感度如“是否含身份证号”、“是否为财务数据”作为数据模型的一等公民存储在 SQLite 的dataresource表中并通过db_relate.py实现策略与数据的双向绑定。这解决了传统 RBAC 中“数据无感知”的根本缺陷。3.1 数据模型设计用 JSONField 存储动态标签兼顾查询效率与灵活性models.py中的DataResource模型是策略决策的核心输入源# sdpmanager_backend/models.py class DataResource(models.Model): name models.CharField(max_length200) description models.TextField() # 标签以 JSON 数组存储支持多值、多层级 tags models.JSONField(defaultlist) # e.g., [sensitive, financial, q3_report] # 敏感等级1-5用于快速过滤高风险资源 sensitivity_level models.PositiveSmallIntegerField( choices[(i, fLevel {i}) for i in range(1, 6)], default3 ) # 关联的业务系统用于策略中的 system_context 条件 system_context models.CharField(max_length100, blankTrue) def __str__(self): return f{self.name} ({self.sensitivity_level}) class Meta: verbose_name Data Resource verbose_name_plural Data ResourcesSQLite 原生不支持 JSON 函数但 Django 3.1 的JSONField通过json_extract和json_each实现了高效查询。例如策略引擎中查询“所有含sensitive标签的资源”-- Django ORM 生成的实际 SQLSQLite SELECT sdpmanager_backend_dataresource.id, sdpmanager_backend_dataresource.name, sdpmanager_backend_dataresource.tags FROM sdpmanager_backend_dataresource WHERE json_extract(sdpmanager_backend_dataresource.tags, $) LIKE %sensitive%;注意json_extract(..., $)返回整个 JSON 字符串LIKE %sensitive%是 SQLite 下最高效的数组包含查询方式比json_each子查询快 3 倍。项目db_relate.py中的get_tagged_resources()函数封装了此逻辑供策略引擎和管理后台复用。3.2 策略与数据的双向绑定db_router.py如何隔离策略库与业务库当系统扩展至多数据库时如策略配置存 PostgreSQL业务数据存 MySQLdb_router.py确保策略读写不污染主业务库。当前项目虽用 SQLite但路由逻辑已预留扩展# sdpmanager_backend/db_router.py class PolicyRouter: route_app_labels {sdpmanager_backend} # 策略相关 app def db_for_read(self, model, **hints): if model._meta.app_label in self.route_app_labels: return policy_db # 策略专用库 return default def db_for_write(self, model, **hints): if model._meta.app_label in self.route_app_labels: return policy_db return default def allow_relation(self, obj1, obj2, **hints): # 允许策略模型与业务模型关联如 DataResource 关联 User if (obj1._meta.app_label in self.route_app_labels or obj2._meta.app_label in self.route_app_labels): return True return None在settings.py中启用DATABASES { default: { # 业务库 ENGINE: django.db.backends.sqlite3, NAME: BASE_DIR / db.sqlite3, }, policy_db: { # 策略库当前指向同一文件生产环境可改为独立 DB ENGINE: django.db.backends.sqlite3, NAME: BASE_DIR / policy_db.sqlite3, } } DATABASE_ROUTERS [sdpmanager_backend.db_router.PolicyRouter]3.2.1 初始化策略库manage.py migrate --databasepolicy_db的实操步骤项目db.sqlite3已预置业务数据但策略库为空。首次运行需手动初始化# 1. 创建策略库迁移文件只针对 sdpmanager_backend app python manage.py makemigrations sdpmanager_backend --databasepolicy_db # 2. 执行迁移创建 policy_db.sqlite3 并建表 python manage.py migrate --databasepolicy_db # 3. 加载初始策略YAML 转数据库记录 python manage.py loaddata sdpmanager_backend/fixtures/initial_policies.json --databasepolicy_db提示initial_policies.json是db_relate.py中export_policies_to_json()生成的确保 YAML 策略与数据库记录一致。若修改 YAML需重新运行loaddata而非migrate——策略是配置数据非 Schema 变更。3.3 审计日志每一次决策都生成可追溯的AccessLog记录零信任的“可验证”体现在审计能力上。log_manager/models.py中的AccessLog模型记录每次决策的完整上下文字段类型说明request_idUUID关联分布式追踪 ID当前用uuid4()生成user_idInteger关联auth_user.id外键可选避免删除用户导致日志断裂ip_addressGenericIPAddressField客户端真实 IP需X-Forwarded-For处理user_agentTextField设备指纹基础信息requested_pathCharField/api/v1/users/http_methodCharFieldGETdecisionCharFieldallow/denymatched_policy_idCharFieldapi_read_sensitive_datadecision_chainJSONField[{subject: user.groups__namedata_analyst, result: true}, ...]timestampDateTimeField自动auto_now_addTrueABACMiddleware.process_view()中的日志写入逻辑# 在 decision 为 False 后追加 if not is_allowed: AccessLog.objects.create( request_idstr(uuid.uuid4()), user_idrequest.user.id if request.user.is_authenticated else None, ip_addressget_client_ip(request), # 自定义函数处理代理头 user_agentrequest.META.get(HTTP_USER_AGENT, ), requested_pathrequest.path, http_methodrequest.method, decisiondeny, matched_policy_idself.policy_engine.last_matched_policy_id, decision_chainself.policy_engine.last_decision_chain # 包含每一步评估详情 ) return HttpResponseForbidden(...)注意get_client_ip()函数在utils.py中实现正确解析X-Forwarded-For和X-Real-IP避免 IP 伪造。项目test.py中的test_ip_extraction单元测试覆盖了 Nginx、Apache 代理场景。4. 开箱即用的调试与验证用内置测试套件和 Admin 界面快速验证策略有效性下载即跑通是毕设项目的生命线。该项目提供了三套验证工具命令行测试脚本、Django Admin 策略管理界面、以及预置的 Postman 集合。它们共同构成一个闭环验证体系让学习者能直观看到“改一行 YAML访问结果立刻变化”。4.1 命令行测试python manage.py test_policy的参数化验证流程test.py中的TestPolicyCommand是一个自定义 Django 管理命令支持交互式策略测试# 测试特定策略对模拟请求的决策结果 python manage.py test_policy \ --policy-id api_read_sensitive_data \ --user-id 1 \ --ip 10.0.1.100 \ --method GET \ --path /api/v1/users/ \ --tags [sensitive] # 输出示例 # Policy api_read_sensitive_data evaluation result: ALLOW # Decision chain: # - Subject user.groups__name: data_analyst: True (user id1 has group data_analyst) # - Condition time_window: True (current time 14:30 is within 09:00-18:00) # - Condition ip_range: True (10.0.1.100 in 10.0.0.0/8)其实现核心是PolicyEngine.simulate_evaluation()方法它接受模拟参数而非真实HttpRequestdef simulate_evaluation(self, policy_id: str, user_id: int, ip: str, method: str, path: str, tags: List[str]): # 1. 加载指定 policy_id 的策略对象 policy self._get_policy_by_id(policy_id) # 2. 构造模拟 request 对象仅含必要属性 mock_request type(MockRequest, (), { user: User.objects.get(iduser_id), META: {REMOTE_ADDR: ip}, method: method, path: path })() # 3. 执行评估返回详细链路 return self._evaluate_with_trace(mock_request, tags)提示test_policy命令默认使用db.sqlite3中的测试数据用户 ID1 属于data_analyst组IP10.0.1.100在白名单内。若要测试拒绝场景将--ip改为192.168.1.100即可触发ip_range条件失败。4.2 Admin 界面策略 YAML 的可视化编辑与实时生效Django Admin 通常只管理数据库但该项目通过admin.py中的PolicyAdmin类将 YAML 策略文件映射为可编辑的 ModelAdmin# sdpmanager_backend/admin.py class PolicyAdmin(admin.ModelAdmin): list_display (policy_id, description, effect, last_modified) readonly_fields (last_modified,) # 自定义保存逻辑将表单数据写入 YAML 文件而非数据库 def save_model(self, request, obj, form, change): policy_file Path(sdpmanager_backend/policies) / f{obj.policy_id}.yaml policy_data { policy_id: obj.policy_id, description: obj.description, subjects: [s.strip() for s in obj.subjects.split(\n) if s.strip()], resources: [{path: obj.resource_path, method: obj.http_method}], conditions: [c.strip() for c in obj.conditions.split(\n) if c.strip()], effect: obj.effect } with open(policy_file, w) as f: yaml.dump(policy_data, f, default_flow_styleFalse, indent2) # 通知 PolicyEngine 重新加载 from sdpmanager_backend.utils.policy_engine import PolicyEngine PolicyEngine().reload_policies() obj.last_modified timezone.now() super().save_model(request, obj, form, change)在 Admin 界面/admin/sdpmanager_backend/policy/中可直接编辑策略的subjects、conditions字段点击保存后YAML 文件实时更新PolicyEngine.reload_policies()清空缓存并重新加载下一个请求即应用新策略。注意PolicyAdmin的list_display中last_modified字段是DateTimeField确保策略变更时间可审计。项目urls.py已将admin/路径包含在urlpatterns中无需额外配置。4.3 Postman 集合预置 12 个场景化测试用例项目根目录下的postman_collection.json导入 Postman 后提供开箱即用的 API 测试集覆盖核心场景用例名操作预期结果技术要点Allow_Sensitive_Read_InOfficeHoursGET/api/v1/users/from10.0.1.100at 14:00200 OK验证ip_rangetime_windowDeny_Sensitive_Read_OutsideHoursGET/api/v1/users/from10.0.1.100at 19:00403 Forbiddentime_window失败Deny_Unauthenticated_AccessGET/api/v1/users/without auth header403 Forbiddensubjects中is_authenticated: true失败Allow_NonSensitive_Read_AnytimeGET/api/v1/public/from any IP200 OKresources.tags不含sensitive跳过策略每个用例的Pre-request Script自动设置Authorization: Token tokenTests脚本验证响应状态码与X-ABAC-Policy-ID响应头// Postman Tests 脚本 pm.test(Status code is 200, function () { pm.response.to.have.status(200); }); pm.test(Policy ID header present, function () { pm.expect(pm.response.headers.get(X-ABAC-Policy-ID)).to.exist; });提示X-ABAC-Policy-ID由ABACMiddleware.process_response()注入便于前端监控策略命中情况。项目views.py中的api_response_wrapper函数统一添加此头。5. 生产就绪的关键改造从 SQLite 快速切换至 MySQL 并启用连接池毕设项目用 SQLite 是为了降低启动门槛但实际部署必须支持 MySQL。项目已预留完整迁移路径且db_router.py和settings.py的数据库配置分离设计让切换只需 5 步无需修改业务代码。5.1 数据库配置切换settings.py中的环境变量驱动模式settings.py使用django-environ解析环境变量MySQL 配置通过.env文件注入# settings.py 第 25 行 import environ env environ.Env() environ.Env.read_env(BASE_DIR / .env) # 读取 .env 文件 DATABASES { default: { ENGINE: django.db.backends.mysql, NAME: env(DB_NAME, defaultsdp_db), USER: env(DB_USER, defaultroot), PASSWORD: env(DB_PASSWORD, default), HOST: env(DB_HOST, default127.0.0.1), PORT: env(DB_PORT, default3306), OPTIONS: { init_command: SET sql_modeSTRICT_TRANS_TABLES, charset: utf8mb4, }, } }.env文件内容示例DB_NAMEsdp_production DB_USERsdp_user DB_PASSWORDStrongPass123! DB_HOSTlocalhost DB_PORT33065.2 MySQL 适配pip install mysqlclient与字符集优化Django 连接 MySQL 需mysqlclient安装命令pip install mysqlclient注意若报错mysql_config not foundUbuntu/Debian 用户需先安装sudo apt-get install python3-dev default-libmysqlclient-dev build-essentialCentOS/RHEL 用户需sudo yum install python3-devel mysql-devel gcc。为避免中文乱码MySQL 服务端需设置utf8mb4字符集。在 MySQL 配置文件my.cnf中添加[client] default-character-set utf8mb4 [mysql] default-character-set utf8mb4 [mysqld] character-set-server utf8mb4 collation-server utf8mb4_unicode_ci重启 MySQL 后创建数据库CREATE DATABASE sdp_production CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;5.3 连接池配置django-db-geventpool提升并发性能SQLite 不支持连接池但 MySQL 在高并发下必须启用。项目requirements.txt已包含django-db-geventpool配置如下# settings.py DATABASES { default: { # ... 其他 MySQL 配置 ENGINE: django_db_geventpool.backends.mysql, OPTIONS: { MAX_CONNS: 20, # 最大连接数 MIN_CONNS: 5, # 最小空闲连接数 RECYCLE: 3600, # 连接复用时间秒 } } }验证连接池是否生效# 启动 Django shell python manage.py shell # 执行查询观察连接数变化 from django.db import connection cursor connection.cursor() cursor.execute(SELECT CONNECTION_ID();) print(cursor.fetchone()) # 返回连接 ID多次执行应看到复用提示django-db-geventpool依赖gevent若项目使用 Gunicorn需添加--worker-class gevent --workers 4 --worker-connections 1000参数。项目sdp_db.py中的test_mysql_connection()函数已封装连接测试逻辑运行python sdp_db.py即可验证 MySQL 连通性与连接池。本文还有配套的精品资源点击获取

Django ABAC动态策略引擎实现零信任访问控制

相关推荐

基于改进粒子群优化的RSSI室内定位：从三边测量到混沌自适应PSO

分布式电源接入配电网承载力评估的Matlab实现与案例分析

WTF-Solidity 极简入门：ERC20 代币标准详解与测试代币发行实战

更多资讯

C++数组完全指南：一维二维、指针与性能优化实战

FFmpeg字幕合并实战：从硬字幕烧录到软字幕封装避坑指南

华为硬件工程师机试实战指南：从信号完整性到单板开发能力图谱

串口服务器里子拆解：从串口转TCP到远程调试，老设备的联网秘密

claude-skills 的 Python CLI 开发实战：基于 Typer、Click、Rich 与 questionary 构建专业命令行工具

es-toolkit/compat 的 takeRight：从数组尾部安全截取元素的 Lodash 兼容实现

今日更新

基于高阶循环谱的MPSK载波频率估计与MATLAB实现

MiniPdf方：.NET开源Office转PDF的高效解决方案

Solon AI v3.10：Java全栈智能体开发框架解析

本周精选

AI驱动的自动化专利分析平台技术解析

LeetCode-Go 题解：1252. Cells with Odd Values in a Matrix（奇数值单元格计数）双解法剖析

在线答疑系统Java毕设实战：Spring Boot前后端分离与状态流转

本月热门

基于YOLO的焊缝缺陷检测：从数据标注到工业部署全流程实践

分布式服务的决策记录

创业产品的首版范围

读完文章，想聊聊您的网站？