一个完整的、生产级别的 WiFi 认证系统通常由两部分组成,您需要分别获取或实现:

(图片来源网络,侵删)
- 网络设备端 (硬件/固件):如路由器、AC(无线控制器)或认证网关,它的作用是拦截未认证用户的网络流量,并将其重定向到指定的认证页面。
- 服务器端 (软件):这是您要的“源码”部分,它是一个 Web 应用,负责提供登录页面、验证用户凭证、记录用户在线状态,并与网络设备进行交互。
下面我将重点讲解服务器端的源码、架构和实现逻辑。
系统架构和工作流程
一个典型的 WiFi 认证系统工作流程如下:
- 用户连接 WiFi:用户的手机或电脑连接到名为
CaptivePortal或类似名称的 WiFi 网络。 - 流量重定向:当用户尝试访问任何网站(
www.baidu.com)时,网络设备(如路由器)会拦截这个 HTTP 请求,并返回一个 HTTP 302 重定向响应,将用户浏览器重定向到认证服务器的登录页面(http://1.1.1.1/login或http://portal.yourdomain.com)。 - 显示登录页:用户的浏览器显示由认证服务器提供的登录页面。
- 用户输入凭证:用户输入用户名和密码,并点击“登录”。
- 服务器验证:认证服务器接收到登录请求,验证用户名和密码。
- 认证成功:
- 服务器在数据库中标记该用户为“已认证”。
- 服务器通过网络设备提供的 API(如 RADIUS 协议、RESTful API 等),通知网络设备为该用户的设备(通过 MAC 地址识别)放行,允许其访问互联网。
- 服务器向用户浏览器返回一个“认证成功”的页面,并可能自动跳转到用户最初想访问的网站。
- 用户上网:用户的设备现在可以正常访问互联网了。
- 下线/超时:用户下线后,网络设备或服务器会检测到其断开连接,并在数据库中清除其在线状态,或者,可以设置一个超时时间(如 8 小时),超时后自动下线。
服务器端源码核心组件与技术选型
技术选型建议
对于初学者或中小型项目,推荐使用现代、高效的 Web 框架:
- 后端:
- Python (Flask/Django): Flask 轻量灵活,适合快速开发 API,Django 功能强大,自带 ORM 和后台管理。
- Node.js (Express): 适合构建高性能的 I/O 密集型应用,API 开发非常方便。
- Go (Gin/Beego): 性能极佳,并发能力强,适合对性能要求高的场景。
- 前端:
简单的 HTML + CSS + JavaScript 即可,为了更好的用户体验,可以加入 Vue.js 或 React。
(图片来源网络,侵删) - 数据库:
- MySQL / PostgreSQL: 功能全面,稳定可靠,适合大多数场景。
- SQLite: 轻量级,适合开发和小型部署。
- Redis: 高性能的键值数据库,非常适合用来存储在线会话、临时认证令牌等。
- 网络交互:
- RADIUS: 这是行业标准协议,绝大多数企业级网络设备都支持,你需要一个 RADIUS 服务器库(如
python-radius)或直接集成 FreeRADIUS。 - RESTful API: 一些现代化的网络设备提供自己的 API,你可以通过 HTTP 请求来控制用户认证状态。
- RADIUS: 这是行业标准协议,绝大多数企业级网络设备都支持,你需要一个 RADIUS 服务器库(如
核心功能模块
- Web 服务器: 提供登录页面、处理登录请求、管理用户会话。
- 用户认证模块: 验证用户名/密码,密码必须加盐哈希存储(如使用
bcrypt或PBKDF2),绝不能明文存储! - 设备管理模块: 记录用户设备的 MAC 地址,并与用户账户关联。
- 网络设备交互模块: 通过 RADIUS 或 API 与网络设备通信,发送认证/下线指令。
- 数据库模块: 存储用户信息、设备信息、认证日志、在线状态等。
- 后台管理模块 (可选): 用于管理员添加/删除用户、查看在线状态、查看流量日志等。
核心代码示例 (使用 Python Flask)
下面是一个使用 Python Flask 框架构建的简化版认证服务器源码,这只是一个演示,展示了核心逻辑,生产环境需要更完善的错误处理、安全加固和功能扩展。
项目结构
wifi_portal/
├── app.py # Flask 主应用
├── config.py # 配置文件
├── requirements.txt # 依赖包
├── templates/
│ ├── base.html # 基础模板
│ └── login.html # 登录页面
└── static/
└── style.css # 样式文件
安装依赖
创建 requirements.txt 文件:
Flask
Flask-SQLAlchemy
Flask-Login
werkzeug # (Flask 依赖,用于密码哈希)
requests # (用于与网络设备API通信)
安装依赖:
pip install -r requirements.txt
config.py - 配置文件
import os
class Config:
SECRET_KEY = os.environ.get('SECRET_KEY') or 'a-very-secret-key'
SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_URL') or 'sqlite:///portal.db'
SQLALCHEMY_TRACK_MODIFICATIONS = False
# 网络设备配置 (以API为例)
PORTAL_API_URL = "http://your-router-api-ip:port/api"
PORTAL_API_TOKEN = "your-secret-api-token"
app.py - Flask 主应用
from flask import Flask, render_template, redirect, url_for, request, flash, session
from flask_sqlalchemy import SQLAlchemy
from flask_login import LoginManager, UserMixin, login_user, logout_user, login_required, current_user
from werkzeug.security import generate_password_hash, check_password_hash
import requests
import os
from config import Config
app = Flask(__name__)
app.config.from_object(Config)
# 初始化扩展
db = SQLAlchemy(app)
login_manager = LoginManager(app)
login_manager.login_view = 'login' # 设置未登录时重定向的视图
# --- 数据库模型 ---
class User(UserMixin, db.Model):
id = db.Column(db.Integer, primary_key=True)
username = db.Column(db.String(80), unique=True, nullable=False)
password_hash = db.Column(db.String(120), nullable=False)
devices = db.relationship('Device', backref='user', lazy=True)
def set_password(self, password):
self.password_hash = generate_password_hash(password)
def check_password(self, password):
return check_password_hash(self.password_hash, password)
class Device(db.Model):
id = db.Column(db.Integer, primary_key=True)
mac_address = db.Column(db.String(17), unique=True, nullable=False)
user_id = db.Column(db.Integer, db.ForeignKey('user.id'), nullable=False)
is_online = db.Column(db.Boolean, default=False)
@login_manager.user_loader
def load_user(user_id):
return User.query.get(int(user_id))
# --- 路由 ---
@app.route('/')
def index():
# 如果用户已登录,显示欢迎页面
if current_user.is_authenticated:
return render_template('index.html', name=current_user.username)
# 否则,重定向到登录页
return redirect(url_for('login'))
@app.route('/login', methods=['GET', 'POST'])
def login():
if current_user.is_authenticated:
return redirect(url_for('index'))
if request.method == 'POST':
username = request.form.get('username')
password = request.form.get('password')
remember = True # 默认记住登录状态
user = User.query.filter_by(username=username).first()
if user and user.check_password(password):
# 登录成功
login_user(user, remember=remember)
# 获取设备的 MAC 地址 (通常从网络设备传递过来,这里简化处理)
# 在实际场景中,网络设备会将用户重定向到 /login?mac=xx:xx:xx:xx:xx:xx
mac_address = request.args.get('mac') or request.form.get('mac')
if mac_address:
# 检查或创建设备记录
device = Device.query.filter_by(mac_address=mac_address).first()
if not device:
device = Device(mac_address=mac_address, user_id=user.id)
db.session.add(device)
device.is_online = True
db.session.commit()
# --- 调用网络设备 API 进行认证 ---
# 这是一个示例,你需要根据你的设备 API 文档来修改
try:
api_url = f"{app.config['PORTAL_API_URL']}/auth"
payload = {
'mac': mac_address,
'token': app.config['PORTAL_API_TOKEN']
}
response = requests.post(api_url, json=payload, timeout=5)
if response.status_code == 200:
print(f"Successfully authenticated device: {mac_address}")
else:
print(f"Failed to authenticate device. API response: {response.text}")
except Exception as e:
print(f"Error calling network device API: {e}")
flash('登录成功!', 'success')
return redirect(url_for('index'))
else:
flash('用户名或密码错误', 'danger')
return render_template('login.html')
@app.route('/logout')
@login_required
def logout():
# 获取当前用户的第一个在线设备 (简化处理)
device = Device.query.filter_by(user_id=current_user.id, is_online=True).first()
if device:
device.is_online = False
db.session.commit()
# --- 调用网络设备 API 进行下线 ---
try:
api_url = f"{app.config['PORTAL_API_URL']}/unauth"
payload = {
'mac': device.mac_address,
'token': app.config['PORTAL_API_TOKEN']
}
requests.post(api_url, json=payload, timeout=5)
print(f"Successfully deauthenticated device: {device.mac_address}")
except Exception as e:
print(f"Error calling network device API for logout: {e}")
logout_user()
flash('您已成功下线!', 'info')
return redirect(url_for('login'))
# 初始化数据库
@app.before_first_request
def create_tables():
db.create_all()
if __name__ == '__main__':
# 确保在运行前设置好环境变量或直接修改 Config
# app.run(debug=True, host='0.0.0.0', port=80)
# 注意:在生产环境中,不应该使用 app.run(),而应使用 Gunicorn/uWSGI 等
app.run(debug=True)
templates/login.html - 登录页面
<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">WiFi 登录</title>
<link rel="stylesheet" href="{{ url_for('static', filename='style.css') }}">
</head>
<body>
<div class="container">
<h1>欢迎使用 WiFi</h1>
{% with messages = get_flashed_messages(with_categories=true) %}
{% if messages %}
{% for category, message in messages %}
<div class="alert alert-{{ category }}">{{ message }}</div>
{% endfor %}
{% endif %}
{% endwith %}
<form method="POST" action="{{ url_for('login') }}">
<div class="form-group">
<label for="username">用户名</label>
<input type="text" id="username" name="username" required>
</div>
<div class="form-group">
<label for="password">密码</label>
<input type="password" id="password" name="password" required>
</div>
<!-- 在实际部署中,MAC 地址通常由网络设备通过重定向 URL 参数传入 -->
<input type="hidden" name="mac" value="{{ request.args.get('mac', '') }}">
<button type="submit" class="btn btn-primary">登录</button>
</form>
</div>
</body>
</html>
templates/index.html - 认证成功页面
<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="UTF-8">认证成功</title>
<link rel="stylesheet" href="{{ url_for('static', filename='style.css') }}">
</head>
<body>
<div class="container">
<h1>欢迎, {{ name }}!</h1>
<p>您已成功连接到互联网。</p>
<a href="{{ url_for('logout') }}" class="btn btn-danger">下线</a>
</div>
</body>
</html>
static/style.css - 样式文件 (可选)
body { font-family: Arial, sans-serif; background-color: #f4f4f4; display: flex; justify-content: center; align-items: center; height: 100vh; margin: 0; }
.container { background: white; padding: 2rem; border-radius: 8px; box-shadow: 0 4px 8px rgba(0,0,0,0.1); text-align: center; width: 100%; max-width: 400px; }
h1 { color: #333; }
.form-group { margin-bottom: 1rem; text-align: left; }
label { display: block; margin-bottom: 0.5rem; }
input[type="text"], input[type="password"] { width: 100%; padding: 0.5rem; border: 1px solid #ddd; border-radius: 4px; box-sizing: border-box; }
.btn { display: inline-block; padding: 0.5rem 1rem; border: none; border-radius: 4px; cursor: pointer; text-decoration: none; }
.btn-primary { background-color: #007bff; color: white; }
.btn-danger { background-color: #dc3545; color: white; }
.alert { padding: 1rem; margin-bottom: 1rem; border: 1px solid transparent; border-radius: 4px; }
.alert-success { color: #155724; background-color: #d4edda; border-color: #c3e6cb; }
.alert-danger { color: #721c24; background-color: #f8d7da; border-color: #f5c6cb; }
.alert-info { color: #0c5460; background-color: #d1ecf1; border-color: #bee5eb; }
如何部署和运行
-
准备环境: 安装 Python 3.x。
(图片来源网络,侵删) -
克隆代码: 将上述代码文件放到一个文件夹中。
-
安装依赖:
pip install -r requirements.txt。 -
配置: 修改
config.py中的SECRET_KEY、数据库 URL 和网络设备 API 信息。 -
初始化数据库: 运行
python app.py一次,它会自动创建portal.db文件,然后停止服务。 -
添加测试用户: 你可以写一个简单的 Python 脚本,或者直接在 SQLite 数据库中手动添加一个用户,并用
werkzeug.security生成密码哈希。from app import app, db, User from werkzeug.security import generate_password_hash with app.app_context(): db.create_all() # 检查用户是否已存在 if not User.query.filter_by(username='admin').first(): admin = User(username='admin') admin.set_password('password123') # 设置密码 db.session.add(admin) db.session.commit() print("用户 'admin' 创建成功!") -
运行服务:
python app.py,默认会在http://127.0.0.1:5000启动。 -
配置网络设备: 这是最关键的一步,你需要登录到你的路由器/AC 的管理后台,找到“Captive Portal”或“Portal 认证”相关选项,将其启用,并设置:
- Portal 类型: 通常是
HTTP Redirect。 - Portal 服务器地址: 设置为你上面运行的服务器的 IP 地址和端口(
http://192.168.1.100:5000)。 - 认证方式: 选择
HTTP或API。
- Portal 类型: 通常是
重要注意事项与生产环境建议
- 安全:
- HTTPS: 必须使用 HTTPS 协议!否则用户的密码和所有流量都会被中间人攻击,可以使用 Let's Encrypt 免费证书。
- 密码安全: 始终使用加盐哈希存储密码。
- CSRF 保护: Flask-WTF 等扩展可以帮你防止跨站请求伪造攻击。
- 输入验证: 对所有用户输入进行严格的验证和清理,防止 SQL 注入等攻击。
- 性能:
- 生产服务器: 不要使用
app.run(),使用 Gunicorn、uWSGI 或 Waitress 等应用服务器来部署。 - 反向代理: 使用 Nginx 或 Apache 作为反向代理,处理静态文件、负载均衡和 SSL 终端。
- 生产服务器: 不要使用
- 功能扩展:
- 用户注册: 添加一个用户注册页面,让用户可以自助注册。
- 付费/时长认证: 集成支付网关(如支付宝、微信支付),实现按时长或流量计费。
- 流量监控: 与网络设备 API 交互,获取用户流量数据并展示。
- 后台管理: 使用 Flask-Admin 或 Django Admin 快速构建一个管理界面。
- 网络设备兼容性: 不同品牌(如 MikroTik, H3C, Huawei, Aruba)的网络设备配置方式差异巨大,请务必查阅你所用设备的官方文档,了解其 Captive Portal 的具体配置方法和 API 接口,RADIUS 是最通用的方案,但配置也最复杂。
这份源码和解释为您提供了一个坚实的基础,您可以根据实际需求和技术栈进行修改和扩展,祝您开发顺利!
