Python3 - Flask+swift实现单点登录

基于 Flask 和 Redis 实现单设备登录的服务端代码和客户端swift、oc代码：

Python flask 实现服务端

from flask import Flask, jsonify, request
from redis import Redisapp = Flask(__name__)
redis_db = Redis()# 用户登录接口，验证用户名和密码，生成并保存 token 到 Redis 中
@app.route('/login', methods=['POST'])
def login():username = request.json.get('username')password = request.json.get('password')# 验证用户身份if check_user(username, password):# 生成 token 并存储到 Redis 中token = generate_token()redis_db.set(token, username)redis_db.expire(token, 1800)  # 设置过期时间为 30 分钟return jsonify({'token': token}), 200return 'Invalid username or password', 401# 业务逻辑接口，需要校验 token 
@app.route('/api/xxx', methods=['GET'])
def api_xxx():token = request.headers.get('Authorization')if not token or not token.startswith('Bearer '):return 'Missing or invalid token', 401# 提取 token 值token = token.split(' ')[1]# 检查 token 是否合法和未过期if redis_db.exists(token):redis_db.expire(token, 1800)  # 续约 token 的过期时间return 'Business logic here', 200return 'Invalid token', 401def check_user(username, password):# TODO: 进行用户身份验证return Truedef generate_token():# TODO: 生成唯一且安全的 tokenreturn 'random_token'if __name__ == '__main__':app.run(debug=True)

在服务端 Python 应用程序中，需要注意以下几点：

1. 通过 Flask-Redis 扩展连接到 Redis 数据库。可以使用 redis.StrictRedis() 或者 redis.Redis.from_url() 方法创建 Redis 实例。
2. 在登录接口和业务逻辑接口中添加 token 的校验代码，如果 token 不合法或已过期，则返回 401 状态码。
3. 在业务逻辑接口中，需要从请求头 Authorization 中提取出 token 值，并检查其是否以 "Bearer " 开头。需要对提取 token 的代码进行安全性检查，防止恶意修改请求头信息。

需要注意的是，上述代码只是一个示例，实际操作时需要进一步完善和优化，例如加入数据验证、错误处理等功能。同时，还需要考虑如何保护敏感数据不被泄露，例如使用 HTTPS 协议、存储密码的哈希值等措施。

客户端为 Swift 语言实现：

import UIKitclass LoginViewController: UIViewController {// 用户名和密码输入框@IBOutlet weak var usernameTextField: UITextField!@IBOutlet weak var passwordTextField: UITextField!override func viewDidLoad() {super.viewDidLoad()}@IBAction func loginButtonTapped(_ sender: Any) {let baseUrl = "https://example.com"let loginUrl = URL(string: "\(baseUrl)/login")!// 创建请求对象var request = URLRequest(url: loginUrl)request.httpMethod = "POST"request.addValue("application/json", forHTTPHeaderField: "Content-Type")// 添加请求体内容let parameters = ["username": usernameTextField.text ?? "","password": passwordTextField.text ?? ""]do {let jsonData = try JSONSerialization.data(withJSONObject: parameters, options: [])request.httpBody = jsonData} catch {print(error.localizedDescription)}// 发送登录请求let task = URLSession.shared.dataTask(with: request) { (data, response, error) inif let error = error {print(error.localizedDescription)return}guard let httpResponse = response as? HTTPURLResponse else {print("Invalid response type")return}if 200...299 ~= httpResponse.statusCode,let data = data,let token = try? JSONSerialization.jsonObject(with: data, options: []) as? [String: String],let authToken = token["token"] {// 登录成功，保存 token 到 Keychain 中self.saveTokenToKeychain(authToken)DispatchQueue.main.async {// 切换到主线程，跳转到下一个页面self.performSegue(withIdentifier: "ShowBusinessLogicSegue", sender: nil)}} else {// 登录失败，显示错误提示DispatchQueue.main.async {let alertController = UIAlertController(title: "Login Failed", message: "Invalid username or password.", preferredStyle: .alert)let okAction = UIAlertAction(title: "OK", style: .default, handler: nil)alertController.addAction(okAction)self.present(alertController, animated: true, completion: nil)}}}task.resume()}func saveTokenToKeychain(_ token: String) {// TODO: 将 token 保存到 Keychain 中}
}class BusinessLogicViewController: UIViewController {override func viewDidLoad() {super.viewDidLoad()}@IBAction func doSomethingButtonTapped(_ sender: Any) {let baseUrl = "https://example.com"let apiUrl = URL(string: "\(baseUrl)/api/xxx")!// 创建请求对象var request = URLRequest(url: apiUrl)request.httpMethod = "GET"// 添加请求头 Authorizationif let authToken = getTokenFromKeychain() {request.addValue("Bearer \(authToken)", forHTTPHeaderField: "Authorization")} else {print("Missing auth token")return}// 发送业务逻辑请求let task = URLSession.shared.dataTask(with: request) { (data, response, error) inif let error = error {print(error.localizedDescription)return}guard let httpResponse = response as? HTTPURLResponse else {print("Invalid response type")return}if 200...299 ~= httpResponse.statusCode {// 处理业务逻辑print("Business logic here")} else if httpResponse.statusCode == 401 {// token 过期或者无效，跳转到登录页面重新登录DispatchQueue.main.async {self.navigationController?.popToRootViewController(animated: true)}} else {// 接口返回错误print("API error")}}task.resume()}func getTokenFromKeychain() -> String? {// TODO: 从 Keychain 中获取 tokenreturn nil}
}

在客户端 Swift 应用程序中，需要注意以下几点：

1. 发送登录请求时，需要将用户名和密码转换为 JSON 格式的字符串，并设置请求头 Content-Type 为 application/json。
2. 在登录成功后，需要将服务器返回的 token 保存到 Keychain 中，以便在下一次请求时使用。
3. 发送业务逻辑请求时，需要在请求头 Authorization 中添加上保存的 token 值（Bearer {token}）。

客户端 Objective-C 实现

在客户端 Objective-C 应用程序中，需要在每次发起请求时添加请求头信息（Authorization），包含存储的 token 值。具体实现可以参考以下示例代码：

// 发送登录请求
NSString *url = @"http://example.com/login"
NSDictionary *parameters = @{@"username": @"your_username", @"password": @"your_password"};
[[AFHTTPSessionManager manager] POST:url parameters:parameters success:^(NSURLSessionDataTask * _Nonnull task, id  _Nullable responseObject) {// 登录成功，将会话标识符保存至本地存储NSString *sessionId = responseObject[@"sessionId"];[[NSUserDefaults standardUserDefaults] setObject:sessionId forKey:@"sessionId"];
} failure:^(NSURLSessionDataTask * _Nullable task, NSError * _Nonnull error) {// 登录失败
}];NSMutableURLRequest *request = [NSMutableURLRequest requestWithURL:[NSURL URLWithString:@"https://example.com/api/xxx"]];
[request setValue:@"Bearer {token}" forHTTPHeaderField:@"Authorization"];
NSURLSessionDataTask *task = [[NSURLSession sharedSession] dataTaskWithRequest:request completionHandler:^(NSData * _Nullable data, NSURLResponse * _Nullable response, NSError * _Nullable error) {// 处理返回数据
}];
[task resume];

需要注意的是，上述代码中 {token} 部分需要替换为从登录接口获取到的有效 token 值。

补充：定期删除redis过期token

要清理过期的 Token，您可以使用 Python 中的定时任务库（例如 APScheduler 或 schedule）来定期检查并删除过期 Token。以下是一个示例：

import datetime
import threadingTOKEN_EXPIRATION_TIME = 1800  # Token 过期时间（秒）
tokens = {}  # 存储 Token 的字典def clear_expired_tokens():"""删除过期的 Token"""now = datetime.datetime.now().timestamp()expired_tokens = [token for token, timestamp in tokens.items() if now - timestamp > TOKEN_EXPIRATION_TIME]for token in expired_tokens:del tokens[token]threading.Timer(TOKEN_EXPIRATION_TIME, clear_expired_tokens).start()def generate_token():"""生成 Token"""token = str(uuid.uuid4())tokens[token] = datetime.datetime.now().timestamp()return token# 启动定时任务
clear_expired_tokens()

在这个示例中，我们首先定义了一个存储 Token 的字典 tokens，以及 Token 的过期时间 TOKEN_EXPIRATION_TIME。然后，我们定义了一个函数 clear_expired_tokens()，该函数会定期检查 tokens 字典中的 Token 是否已过期，并将过期的 Token 从字典中删除。

接下来，我们定义了一个生成 Token 的函数 generate_token()，该函数生成随机的 UUID 并将其作为键存储到 tokens 字典中，并将当前时间戳作为值存储。最后，我们使用 threading.Timer 启动一个定时任务，以便每隔 TOKEN_EXPIRATION_TIME 秒调用一次 clear_expired_tokens() 函数。

请注意，这只是一个示例实现，您可以根据自己的需求进行修改和优化。例如，如果您的应用程序使用数据库存储 Token，则可以考虑在数据库层面上清理过期 Token。