手把手教你搭建档案防火墙管理系统实战指南

环境准备与依赖安装

在开始构建档案防火墙管理系统之前，我们需要准备标准的Python开发环境。本系统基于Python 3.8+开发，利用Flask作为轻量级Web框架，SQLite作为轻量级数据库，确保零门槛快速落地。

请在终端中执行以下命令创建项目目录并初始化虚拟环境。虚拟环境能够隔离项目依赖，避免版本冲突，这是开发的第一步。

 创建项目文件夹
mkdir archive_firewall_system
cd archive_firewall_system
创建虚拟环境
python3 -m venv venv
激活虚拟环境 (Linux/Mac)
source venv/bin/activate
激活虚拟环境 (Windows)
venv\Scripts\activate
安装核心依赖
pip install flask werkzeug

注意： werkzeug 用于处理安全的文件名和密码哈希，是Flask的核心依赖，这里显式安装以确保版本一致。安装完成后，我们将进入项目结构搭建阶段。

项目目录结构规划

为了保证代码的可维护性，我们采用模块化的目录结构。请在项目根目录下执行以下命令创建必要的文件夹和文件：

mkdir -p app/templates app/static/uploads app/static/css
touch app/__init__.py app/config.py app/models.py app/firewall.py app/views.py app/database.py
touch app/templates/index.html app/templates/login.html
touch run.py

最终的目录结构将包含配置文件、数据库模型、核心防火墙逻辑、视图控制器以及前端模板。这种结构清晰地将业务逻辑与安全控制分离开来。

核心配置文件编写

配置文件 app/config.py 定义了系统的运行参数，包括上传文件的大小限制、允许的扩展名以及密钥配置。请将以下代码完整复制到该文件中：

import os
基础目录配置
BASE_DIR = os.path.abspath(os.path.dirname(__file__))
密钥配置，用于Session加密
SECRET_KEY = 'change_this_to_a_random_string_in_production'
上传文件配置
UPLOAD_FOLDER = os.path.join(BASE_DIR, 'static/uploads')
ALLOWED_EXTENSIONS = {'txt', 'pdf', 'png', 'jpg', 'jpeg', 'doc', 'docx'}
MAX_CONTENT_LENGTH = 16  1024  1024   限制上传最大为16MB
数据库配置
SQLALCHEMY_DATABASE_URI = 'sqlite:///' + os.path.join(BASE_DIR, 'data.db')
SQLALCHEMY_TRACK_MODIFICATIONS = False

这里我们设定了严格的文件类型白名单和大小限制，这是档案防火墙的第一道防线，防止恶意文件上传或服务器资源耗尽。

数据库模型设计

我们需要设计三个核心表：用户表、档案表和访问日志表。编辑 app/models.py，输入以下代码：

from flask_sqlalchemy import SQLAlchemy
from datetime import datetime
db = SQLAlchemy()
class User(db.Model):
id = db.Column(db.Integer, primary_key=True)
username = db.Column(db.String(80), unique=True, nullable=False)
password = db.Column(db.String(120), nullable=False)
role = db.Column(db.String(20), nullable=False)   admin, user, auditor
class Archive(db.Model):
id = db.Column(db.Integer, primary_key=True)
filename = db.Column(db.String(120), nullable=False)
filepath = db.Column(db.String(200), nullable=False)
sensitivity = db.Column(db.String(20), nullable=False)  public, internal, secret
uploader_id = db.Column(db.Integer, db.ForeignKey('user.id'))
upload_time = db.Column(db.DateTime, default=datetime.utcnow)
class AccessLog(db.Model):
id = db.Column(db.Integer, primary_key=True)
user_id = db.Column(db.Integer, db.ForeignKey('user.id'))
archive_id = db.Column(db.Integer, db.ForeignKey('archive.id'))
action = db.Column(db.String(50))  upload, download, access_denied
ip_address = db.Column(db.String(50))
timestamp = db.Column(db.DateTime, default=datetime.utcnow)
details = db.Column(db.String(200))

重点说明： sensitivity 字段用于标记档案的敏感级别，这是防火墙进行访问控制的关键依据；AccessLog 表则记录所有操作，满足审计合规要求。

数据库初始化脚本

为了方便测试，我们需要一个脚本来初始化数据库并创建一个默认管理员账号。编辑 app/database.py：

from app import create_app
from app.models import db, User
from werkzeug.security import generate_password_hash
app = create_app()
with app.app_context():
db.create_all()
检查是否已存在管理员，若不存在则创建
if not User.query.filter_by(username='admin').first():
admin = User(username='admin', password=generate_password_hash('admin123'), role='admin')
db.session.add(admin)
db.session.commit()
print("数据库初始化完成，管理员账号已创建: admin / admin123")
else:
print("数据库已存在。")

防火墙核心逻辑实现

这是本系统的核心，即 app/firewall.py。我们将实现一个基于角色的访问控制（RBAC）和敏感级别匹配的类。任何档案访问请求都必须经过此类的验证。

from app.models import User, Archive, AccessLog, db
from flask import session, request
class ArchiveFirewall:
定义权限等级映射
ROLE_LEVELS = {'admin': 3, 'auditor': 2, 'user': 1}
SENSITIVITY_LEVELS = {'public': 1, 'internal': 2, 'secret': 3}
@staticmethod
def check_access(user_id, archive_id):
"""
检查用户是否有权限访问指定档案
返回: (bool, str) -> (是否允许, 拒绝原因)
"""
user = User.query.get(user_id)
archive = Archive.query.get(archive_id)
if not user or not archive:
return False, "资源不存在"
user_level = ArchiveFirewall.ROLE_LEVELS.get(user.role, 0)
file_level = ArchiveFirewall.SENSITIVITY_LEVELS.get(archive.sensitivity, 99)
核心判断逻辑：用户等级必须大于等于档案敏感等级
if user_level >= file_level:
ArchiveFirewall.log_action(user, archive, "access_granted")
return True, "允许访问"
else:
reason = f"权限不足: 用户级别{user.role}低于档案级别{archive.sensitivity}"
ArchiveFirewall.log_action(user, archive, "access_denied", details=reason)
return False, reason
@staticmethod
def log_action(user, archive, action, details=None):
"""记录所有访问日志"""
log = AccessLog(
user_id=user.id,
archive_id=archive.id,
action=action,
ip_address=request.remote_addr,
details=details
)
db.session.add(log)
db.session.commit()

这段代码实现了严格的等级检查。例如，普通用户（level 1）试图下载机密档案（level 3）时，系统会自动拦截并记录日志。

后端视图与路由开发

接下来编写 app/views.py，处理HTTP请求。这里包含登录、上传、下载等核心接口，并在下载接口中调用防火墙逻辑。

from flask import Blueprint, render_template, request, redirect, url_for, flash, send_from_directory, session
from werkzeug.security import check_password_hash
from werkzeug.utils import secure_filename
from app.models import db, User, Archive
from app.firewall import ArchiveFirewall
from app.config import ALLOWED_EXTENSIONS, UPLOAD_FOLDER
import os
bp = Blueprint('main', __name__)
def allowed_file(filename):
return '.' in filename and filename.rsplit('.', 1)[1].lower() in ALLOWED_EXTENSIONS
@bp.route('/')
def index():
if 'user_id' not in session:
return redirect(url_for('main.login'))
archives = Archive.query.all()
return render_template('index.html', archives=archives, user_role=session.get('role'))
@bp.route('/login', methods=['GET', 'POST'])
def login():
if request.method == 'POST':
username = request.form['username']
password = request.form['password']
user = User.query.filter_by(username=username).first()
if user and check_password_hash(user.password, password):
session['user_id'] = user.id
session['role'] = user.role
return redirect(url_for('main.index'))
flash('登录失败，请检查用户名或密码')
return render_template('login.html')
@bp.route('/upload', methods=['POST'])
def upload():
if 'user_id' not in session:
return redirect(url_for('main.login'))
if 'file' not in request.files:
flash('没有文件部分')
return redirect(url_for('main.index'))
file = request.files['file']
sensitivity = request.form.get('sensitivity', 'public')
if file.filename == '':
flash('未选择文件')
return redirect(url_for('main.index'))
if file and allowed_file(file.filename):
filename = secure_filename(file.filename)
filepath = os.path.join(UPLOAD_FOLDER, filename)
file.save(filepath)
new_archive = Archive(filename=filename, filepath=filepath, sensitivity=sensitivity, uploader_id=session['user_id'])
db.session.add(new_archive)
db.session.commit()
flash('档案上传成功')
else:
flash('不支持的文件类型')
return redirect(url_for('main.index'))
@bp.route('/download/')
def download(archive_id):
if 'user_id' not in session:
return redirect(url_for('main.login'))
防火墙检查点
allowed, reason = ArchiveFirewall.check_access(session['user_id'], archive_id)
if allowed:
archive = Archive.query.get(archive_id)
return send_from_directory(os.path.dirname(archive.filepath), os.path.basename(archive.filepath), as_attachment=True)
else:
flash(f'访问被拒绝: {reason}')
return redirect(url_for('main.index'))
@bp.route('/logout')
def logout():
session.clear()
return redirect(url_for('main.login'))

在 download 函数中，我们在读取文件系统之前强制调用了 ArchiveFirewall.check_access，确保没有任何文件能绕过检查被下载。

应用初始化与入口文件

编辑 app/__init__.py 来组装应用：

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from app.config import SECRET_KEY, SQLALCHEMY_DATABASE_URI, SQLALCHEMY_TRACK_MODIFICATIONS, UPLOAD_FOLDER
import os
def create_app():
app = Flask(__name__)
app.config.from_object('app.config')
确保上传目录存在
if not os.path.exists(UPLOAD_FOLDER):
os.makedirs(UPLOAD_FOLDER)
from app.models import db
db.init_app(app)
from app.views import bp
app.register_blueprint(bp)
return app

编辑根目录下的 run.py 作为启动入口：

from app import create_app
app = create_app()
if __name__ == '__main__':
app.run(debug=True, port=5000)

前端页面开发

我们需要一个简单的登录页和管理首页。首先编辑 app/templates/login.html：

系统登录
{% with messages = get_flashed_messages() %}
{% if messages %}
{{ messages[0] }}
{% endif %}
{% endwith %}

用户名: 
密码: 
登录

编辑 app/templates/index.html：

档案管理控制台 (当前角色: {{ user_role }})
退出登录

上传新档案



公开
内部
机密

上传


档案列表
{% with messages = get_flashed_messages() %}
{% if messages %}
{{ messages[0] }}
{% endif %}
{% endwith %}

{% for archive in archives %}

{% endfor %}

ID
文件名
敏感级别
操作

{{ archive.id }}
{{ archive.filename }}
{{ archive.sensitivity }}
下载

系统运行与功能验证

现在所有代码已就绪，我们将启动系统并进行防火墙功能的实际验证。

步骤1：初始化数据库

在终端执行以下命令，这会创建SQLite数据库文件和默认管理员账号。

python app/database.py

终端应显示：“数据库初始化完成，管理员账号已创建: admin / admin123”。

步骤2：启动服务

python run.py

服务将在 http://127.0.0.1:5000 启动。