ToB企服应用市场:ToB评测及商务社交产业平台

标题: 网络安全-怎样设计一个安全的API（安全角度） [打印本页]

---

作者: 曂沅仴駦    时间: 2024-7-12 01:12
标题: 网络安全-怎样设计一个安全的API（安全角度）

---

API安全概述

使用API可进行以下常见的攻击：

• 注入攻击（SQL注入、下令注入、XSS等）
  
• DOS/DDOS攻击
  
• SSRF
  
• 未授权/水平（垂直）越权
  
• 敏感数据泄露
  
• 中间人攻击
  
• 更改请求方法调用
  
• 并发攻击
  
• 重放攻击
  
• 数据篡改和伪造
  

有以下常见的防御方式：

• 资源请求限制，通过限频等手段来解决DOS、DDOS攻击
  
• 线程加锁来解决并发攻击
  
• 权限控制，通过ABAC、RBAC等方式解决越权攻击
  
• 敏感信息防泄露，通太过类分级引擎，数据库加密存储等方式来解决敏感信息泄露
  
• 防重放，通过API认证解决重放攻击
  
• 加密，例如HTTPS来解决中间人攻击
  
• 安全产物，例如API网关、WAF等来解决大部分攻击
  

当然，有些还是必要API后端代码来进行防御，例如下令注入、SSRF等。
本文以API身份认证为主要内容，浅谈各种认证的使用场景与优缺点，同时穿插部分攻击与防御。
设计一个安全的API

一个基本的API

主要代码

1. import uuid
2. import re
3. import traceback
5. from flask import Flask, request, jsonify
6. from mysql import MysqlCli
7. from log import log
8. from setting import *
10. app = Flask(__name__)
11. log.set_file()
13. # 验证username
14. def validate_username(username:str)->bool:
15.     if len(username) > 20:
16.         return False
17.     return True
19. # 验证手机号
20. def validate_phone_number(phone_number:str)->bool:
21.     # 使用正则表达式检查手机号格式
22.     pattern = re.compile(r'^1[3456789]\d{9}$')
23.     if re.match(pattern, phone_number):
24.         return True
25.     else:
26.         return False
28. # 管理员注册用户接口
29. @app.route('/api/v1.0/admin/add_user', methods=['POST'])
30. def add_user():
31.     resp = {
32.         "requestid": uuid.uuid4()
33.     }
34.     if request.is_json:
35.         data = request.get_json()
36.         username = data.get('username')
37.         if username is None:
38.             resp['error'] = 'username is required'
39.             log.logger.error(f"url:{request.url},params:{username},resp:{resp}")
40.             return jsonify(resp),400
41.         phone_number = data.get('phone_number')
42.         if phone_number is None:
43.             resp['error'] = 'phone_number is required'
44.             log.logger.error(f"url:{request.url},params:{phone_number},resp:{resp}")
45.             return jsonify(resp),400
47.         if not validate_username(username):
48.             resp['error'] = 'username length should not exceed 10'
49.             log.logger.error(f"url:{request.url},params:{username},resp:{resp}")
50.             return jsonify(resp),400
51.         try:
52.             cli = MysqlCli(MYSQL_HOST,MYSQL_USER,MYSQL_PASSWORD,MYSQL_DATABASE)
53.             cli.insert_one("users",
54.                 {
55.                     "username": username,
56.                     "phone_number": phone_number
57.                 }
58.             )
59.             cli.close()
60.             resp['message'] = f'success to add user:{username}!'
61.             log.logger.info(f"url:{request.url},params:{username},resp:{resp}")
62.             return jsonify(resp)
63.         except Exception:
64.             log.logger.info(f"url:{request.url},params:{username},resp:{resp}")
65.             resp['error'] = f"failed to insert to mysql:{traceback.format_exc()}"
66.             return jsonify(resp),500
67.     else:
68.         resp['error'] = "Invalid JSON format in request"
69.         return jsonify(resp),400
71. # 用户获取信息接口
72. @app.route('/api/v1.0/get_user_info', methods=['POST'])
73. def get_user_info():
74.     resp = {
75.         "requestid": uuid.uuid4()
76.     }
77.     if request.is_json:
78.         data = request.get_json()
79.         username = data.get('username')
80.         if username is None:
81.             resp['error'] = 'username is required'
82.             log.logger.error(f"url:{request.url},params:{username},resp:{resp}")
83.             return jsonify(resp),400
85.         if not validate_username(username):
86.             resp['error'] = 'username length should not exceed 10'
87.             log.logger.error(f"url:{request.url},params:{username},resp:{resp}")
88.             return jsonify(resp),400
89.         try:
90.             cli = MysqlCli(MYSQL_HOST, MYSQL_USER, MYSQL_PASSWORD, MYSQL_DATABASE)
91.             sql = f"select * from users where username = '{username}' limit 1"
92.             user = cli.select_all(sql)
93.             resp['message'] = f'success to get user:{user}.'
94.             log.logger.info(f"url:{request.url},params:{username},resp:{resp}")
95.             return jsonify(resp)
96.         except Exception:
97.             resp['error'] = f'failed to get user, error:{traceback.format_exc()}'
98.             log.logger.error(f"url:{request.url},params:{username},resp:{resp}")
99.             return jsonify(resp),500
100.     else:
101.         resp['error'] = "Invalid JSON format in request"
102.         return jsonify(resp),400
105. if __name__ == '__main__':
106.     app.run()

复制代码

可以看到共有两个API，管理员注册用户接口、用户获取信息接口，拥有以下功能或安全步伐：

• 版本控制
  
• 日记记载
  
• 请求方法校验
  
• 请求数据校验
  

调用

API的一些问题

• 没有身份认证，只要有人知道api地址、方法、参数就能调用
  
• 没有调用次数限制，可能被Dos或DDos攻击，造成服务器压力
  
• 没有防止数据篡改或伪造，被抓包后可能被篡改或伪造数据
  
• 没有对重放攻击的防御，抓包后可能重放包
  
• 越权问题，管理员接口任何人都可以调用
  
• SQL注入，没有做参数化查询等防止SQL操纵，可能被拖库
  

BasicAuth

认证流程

• 客户端发送请求头Authorization为Basic username:password（base64编码）的数据包
  
• 服务端对请求头Authorization判断，解码后从数据库查询判断账号密码是否正确
  

主要代码

1. def basic_auth(f):
2.     @wraps(f)
3.     def decorated_function(*args, **kwargs):
4.         resp = {
5.             "requestid": uuid.uuid4()
6.         }
7.         auth_header = request.headers.get('Authorization')
8.         if not auth_header or not auth_header.startswith('Basic '):
9.             resp['error'] = 'basic auth is required'
10.             log.logger.error(f"url:{request.url},resp:{resp}")
11.             return jsonify(resp), 400
12.         else:
13.             try:
14.                 encoded_credentials = auth_header.split(' ')[1]
15.                 decoded_credentials = base64.b64decode(encoded_credentials).decode('utf-8')
16.                 username, password = decoded_credentials.split(':')
17.                 if not check_basic_auth(username, password):
18.                     resp['error'] = 'basic auth failed,check your username or password is right'
19.                     log.logger.error(f"url:{request.url},resp:{resp}")
20.                     return jsonify(resp), 401
21.             except Exception:
22.                 resp['error'] = f'basic auth failed,err: {traceback.format_exc()}'
23.                 log.logger.error(f"url:{request.url},resp:{resp}")
24.                 return jsonify(resp), 500
26.         return f(*args, **kwargs)
28.     return decorated_function
29. # 管理员注册用户接口 v2.0 增加密码
30. @app.route('/api/v2.0/admin/basic_auth/add_user', methods=['POST'])
31. @basic_auth
32. def add_user_basic_auth():
33.     resp = {
34.         "requestid": uuid.uuid4()
35.     }
36.     # 请求是json格式
37.     if request.is_json:
38.         data = request.get_json()
39.         username = data.get('username')
40.         # username检查
41.         if username is None:
42.             resp['error'] = 'username is required'
43.             log.logger.error(f"url:{request.url},params:{username},resp:{resp}")
44.             return jsonify(resp),400
45.         if not validate_username(username):
46.             resp['error'] = 'username length should not exceed 10'
47.             log.logger.error(f"url:{request.url},params:{username},resp:{resp}")
48.             return jsonify(resp),400
49.         # phone_number检查
50.         phone_number = data.get('phone_number')
51.         if phone_number is None:
52.             resp['error'] = 'phone_number is required'
53.             log.logger.error(f"url:{request.url},params:{phone_number},resp:{resp}")
54.             return jsonify(resp),400
55.         # 插入数据库
56.         try:
57.             cli = MysqlCli(MYSQL_HOST,MYSQL_USER,MYSQL_PASSWORD,MYSQL_DATABASE)
58.             cli.insert_one("users",
59.                 {
60.                     "username": username,
61.                     "phone_number": phone_number
62.                 }
63.             )
64.             pwd = generate_random_password(secrets.choice(range(8, 17)))
65.             cli.insert_one("passwords", {
66.                 "username": username,
67.                 "password": pwd
68.             })
69.             cli.close()
70.             resp['message'] = f'success to add user:{username},password {pwd},remember it!'
71.             log.logger.info(f"url:{request.url},params:{username},resp:{resp}")
72.         # 异常返回
73.         except Exception:
74.             log.logger.info(f"url:{request.url},params:{username},resp:{resp}")
75.             resp['error'] = f"failed to insert to mysql:{traceback.format_exc()}",500
76.             return jsonify(resp)
77.         return jsonify(resp)
78.     # 请求不是json格式
79.     else:
80.         resp['error'] = "Invalid JSON format in request"
81.         return jsonify(resp),400

复制代码

问题

• 没有调用次数限制，可能被Dos或DDos攻击，造成服务器压力
  
• 没有防止数据篡改或伪造，被抓包后可能被篡改或伪造数据
  
• 没有对重放攻击的防御，抓包后可能重放包
  
• 越权问题，管理员接口任何人都可以调用
  
• SQL注入，没有做参数化查询等防止SQL操纵，可能被拖库
  

引入问题：

• 请求的密码进行base64编码，轻易获得并解码获得明文
  
• 数据库密码明文存储
  
• 数据库sql语句执行没有按事件处理，可能用户插入到数据库但密码没有入库
  

API Key

流程

• 客户端发送请求时通过query string、请求头（X-API-Key或其他自定义请求头）、Cookie中携带apikey
  
• 服务端根据约定的方式获取后查询数据库判断是否存在
  

主要代码

1. def api_key_auth(f):
2.     @wraps(f)
3.     def decorated_function(*args, **kwargs):
4.         resp = {
5.             "requestid": uuid.uuid4()
6.         }
7.         # api key检查
8.         if "api_key" not in request.headers:
9.             resp['error'] = 'api_key is required in headers'
10.             log.logger.error(f"url:{request.url},resp:{resp}")
11.             return jsonify(resp), 400
12.         if not check_api_key(request.headers["api_key"]):
13.             resp['error'] = f'api_key {request.headers["api_key"]} is invalid'
14.             log.logger.error(f"url:{request.url},resp:{resp}")
15.             return jsonify(resp), 401
16.         return f(*args, **kwargs)
18.     return decorated_function
19. # 管理员注册用户接口 v2.0 增加api token
20. @app.route('/api/v2.0/admin/api_key/add_user', methods=['POST'])
21. @api_key_auth
22. def add_user_api_key():
23.     resp = {
24.         "requestid": uuid.uuid4()
25.     }
26.     # request是json
27.     if request.is_json:
28.         data = request.get_json()
29.         username = data.get('username')
30.         if username is None:
31.             resp['error'] = 'username is required'
32.             log.logger.error(f"url:{request.url},params:{username},resp:{resp}")
33.             return jsonify(resp),400
34.         phone_number = data.get('phone_number')
35.         if phone_number is None:
36.             resp['error'] = 'phone_number is required'
37.             log.logger.error(f"url:{request.url},params:{phone_number},resp:{resp}")
38.             return jsonify(resp),400
39.         if not validate_username(username):
40.             resp['error'] = 'username length should not exceed 10'
41.             log.logger.error(f"url:{request.url},params:{username},resp:{resp}")
42.             return jsonify(resp),400
43.         try:
44.             cli = MysqlCli(MYSQL_HOST,MYSQL_USER,MYSQL_PASSWORD,MYSQL_DATABASE)
45.             cli.insert_one("users",
46.                 {
47.                     "username": username,
48.                     "phone_number": phone_number
49.                 }
50.             )
51.             key = secrets.token_hex(16)
52.             cli.insert_one("keys",
53.                            {
54.                                "username": username,
55.                                "key": key
56.                            })
57.             cli.close()
58.             resp['message'] =f'success to add user:{username},key {key},remember it!'
59.             log.logger.info(f"url:{request.url},params:{username},resp:{resp}")
60.         except Exception:
61.             log.logger.info(f"url:{request.url},params:{username},resp:{resp}")
62.             resp['error'] = f"failed to insert to mysql:{traceback.format_exc()}"
63.             return jsonify(resp),500
64.         return jsonify(resp)
65.     else:
66.         resp['error'] = "Invalid JSON format in request"
67.         return jsonify(resp),400

复制代码

博主这里实现时，添加了自定义请求头api_key

问题

• 没有调用次数限制，可能被Dos或DDos攻击，造成服务器压力
  
• 没有防止数据篡改或伪造，被抓包后可能被篡改或伪造数据
  
• 没有对重放攻击的防御，抓包后可能重放包
  
• 越权问题，管理员接口任何人都可以调用
  
• SQL注入，没有做参数化查询等防止SQL操纵，可能被拖库
  

引入问题：

• API key明文发送，容器抓包获取
  
• API key明文存储
  
• 数据库sql语句执行没有按事件处理，可能用户插入到数据库但api key没有入库
  

Bearer auth/Token auth

流程

• 请求端通过某种认证机制（好比用户名密码登录、OAuth 认证等）获取令牌。 在发起 HTTP 请求时，客户端将这个令牌添加到Authorization 请求头中，格式为 “Bearer token”。
  
• 服务器接收到请求后，会验证这个令牌的有效性，假如有效则允许请求继续处理，否则拒绝访问。
  

使用 Bearer authentication 的优势在于令牌本身可以包罗更多的信息、具有较长的有效期，并且不必要在服务器端保存会话状态，这样可以减轻服务器负担并提高安全性。
这里就不实现了，后面通过jwt token，算是实现其中的一种。
Digest Auth

流程

• 客户端发送一个未经认证的请求给服务器。 服务器返回一个 401 Unauthorized响应，有一个响应头WWW-Authenticate，其中包罗一个随机数（nonce）和其他认证必要的信息。
  
• 客户端收到 401响应后，会向用户提示输入用户名和密码，然后根据特定的算法（通常是 MD5）对用户名、密码、随机数（nonce）、HTTP 方法和请求的URI 进行摘要盘算。
  
• 客户端将盘算出的摘要放在 Authorization 请求头中发送给服务器。
  
• 服务器收到请求后，会根据事先约定好的算法再次盘算摘要，假如两个摘要匹配，则验证通过，否则拒绝访问。
  

涉及的几个常见参数如下：

• realm，必选。是一个掩护空间的名称，用于向用户表明请求的资源属于哪个掩护空间。它通常用于表现一组受掩护的资源，用于构造摘要字符串。
  
• nonce，必选。是一个唯一的字符串，401时由服务器生成并发送给客户端。它用于防止重放攻击（replay attack）。每次认证请求都会使用一个新的 nonce 值，使得每次请求的摘要都是不同的，从而提高了安全性。
  
• qop(Quality of Protection) ，必选。可以是 “auth” 或 “auth-int”。auth 代表身份验证，而 auth-int 代表身份验证和消息完整性掩护。Qop 的存在使得摘要认证更加灵活和安全。
  
• algorithm，可选，默认MD5。指定了用于盘算摘要的哈希算法，通常是 MD5。服务器在 WWW-Authenticate 响应头中指定，客户端按照这个算法进行摘要盘算。还有MD5-sess、SHA、SHA-256、SHA-512等。
  
• nc，可选，是一个计数器，用于跟踪特定 nonce 值的使用次数。每次客户端发送请求时，Nc 都会递增，帮助防止重放攻击。
  
• cnonce(Client Nonce) ，可选。是客户端生成的随机字符串，用于与服务器的 nonce 一起使用，以增加请求的独特性和安全性。
  
• opaque，可选。是服务器生成的字符串，客户端在后续请求中必须原样返回。它用来保持服务器状态或防止某些类型的攻击。
  
• charset，可选。默认utf-8，编码方式。
  
• userhash，可选。默认false。服务端返回的是否支持username哈希。
  

主要代码

1. # 将字符串保存到Redis中，并设置过期时间
2. def save_nonce_with_expiry(key, value, expiry_seconds):
3.     """
4.     :param key: 键
5.     :param value: 值
6.     :param expiry_seconds: 过期时间（秒）
7.     """
8.     # 连接Redis数据库
9.     redis_client = redis.StrictRedis(host=REDIS_HOST,password=REDIS_PASSWORD, port=REDIS_PORT, db=REDIS_DB)
10.     redis_client.setex(key, expiry_seconds, value)
12. def check_nonce(key):
13.     """
14.     检查字符串是否存在于Redis中
15.     :param key: 键
16.     :return: 布尔值，表示键是否存在
17.     """
18.     # 连接Redis数据库
19.     redis_client = redis.StrictRedis(host=REDIS_HOST,password=REDIS_PASSWORD ,port=REDIS_PORT, db=REDIS_DB)
20.     return redis_client.exists(key)
22. # 校验username 返回密码
23. def check_username(username):
24.     try:
25.         cli = MysqlCli(MYSQL_HOST, MYSQL_USER, MYSQL_PASSWORD, MYSQL_DATABASE)
26.         sql = f"select * from passwords where username = '{username}' limit 1"
27.         user = cli.select_one(sql)
28.         return user['password'] if user else ''
29.     except Exception:
30.         log.logger.error(f"check_basic_auth failed,error:{traceback.format_exc()}")
31.         return False
33. # 校验response
34. def check_response(response,realm,username,password,method,uri,req_nonce,nc,cnonce,qop):
35.     try:
36.         log.logger.info(f"check_response ,response:{response},realm:{realm},username:{username},"
37.                         f"password:{password},method:{method},uri:{uri},req_nonce:{req_nonce},nc:{nc},"
38.                         f"cnonce:{cnonce},qop:{qop}")
39.         # 校验nonce
40.         if not check_nonce(req_nonce):
41.             log.logger.error(f"check_signature failed,error:{req_nonce} not exist!")
42.             return False
43.         ha1=hashlib.md5(f"{username}:{realm}:{password}".encode()).hexdigest()
44.         ha2=hashlib.md5(f"{method}:{uri}".encode()).hexdigest()
45.         return response == hashlib.md5(f"{ha1}:{req_nonce}:{nc}:{cnonce}:{qop}:{ha2}".encode()).hexdigest()
46.     except Exception:
47.         log.logger.error(f"check_responce failed,error:{traceback.format_exc()}")
48.         return False
50. # 添加用户 用于添加用户时无需额外增加用户关联的密钥等信息时的接口
51. def add_user():
52.     resp = {
53.         "requestid": uuid.uuid4()
54.     }
55.     log.logger.info(request.headers)
57.     # request是json
58.     if request.is_json:
59.         data = request.get_json()
60.         username = data.get('username')
61.         if username is None:
62.             resp['error'] = 'username is required'
63.             log.logger.error(f"url:{request.url},params:{username},resp:{resp}")
64.             return jsonify(resp)
65.         phone_number = data.get('phone_number')
66.         if phone_number is None:
67.             resp['error'] = 'phone_number is required'
68.             log.logger.error(f"url:{request.url},params:{phone_number},resp:{resp}")
69.             return jsonify(resp)
71.         if not validate_username(username):
72.             resp['error'] = 'username length should not exceed 10'
73.             log.logger.error(f"url:{request.url},params:{username},resp:{resp}")
74.             return jsonify(resp)
75.         try:
76.             cli = MysqlCli(MYSQL_HOST, MYSQL_USER, MYSQL_PASSWORD, MYSQL_DATABASE)
77.             cli.insert_one("users",
78.                            {
79.                                "username": username,
80.                                "phone_number": phone_number
81.                            }
82.                            )
83.             cli.close()
84.             resp['message'] = f'success to add user:{username}!'
85.             log.logger.info(f"url:{request.url},params:{username},resp:{resp}")
86.         except Exception:
87.             log.logger.info(f"url:{request.url},params:{username},resp:{resp}")
88.             resp['error'] = f"failed to insert to mysql:{traceback.format_exc()}"
89.         return jsonify(resp)
90.     else:
91.         resp['error'] = "Invalid JSON format in request"
92.         return jsonify(resp)
94. # 管理员注册用户接口 v2.0增加digest算法
95. @app.route('/api/v2.0/admin/digest_auth/add_user', methods=['POST'])
96. @digest_auth
97. def add_user_digest():
98.     return add_user()

复制代码

输入账号密码后登录，发起请求

问题

• 没有调用次数限制，可能被Dos或DDos攻击，造成服务器压力
  
• 没有防止数据篡改或伪造，被抓包后可能被篡改或伪造数据
  
• 没有对重放攻击的防御，抓包后可能重放包
  
• 越权问题，管理员接口任何人都可以调用
  
• SQL注入，没有做参数化查询等防止SQL操纵，可能被拖库
  

优点：

• 可以防止重放攻击
  

引入问题：

• 发送两个请求，更加斲丧资源
  
• 必要存储nonce，这里有设置逾期时间
  
• 实现比较复杂
  

JWT Token

流程

通常由三个部分构成：header、payload 和 signature。

• header：包罗两个部分：令牌类型（即 “JWT”）和所使用的签名算法（如 HMAC SHA256 或 RSA）。
  
• payload：包罗声明（claims）。声明是关于实体（通常是用户）和其他数据的声明。
  
  
  
  • iss（Issuer）：该声明标识了 JWT 的发行者。
    
  • sub（Subject）：该声明标识了 JWT 的主题，即所形貌的实体。
    
  • aud（Audience）：该声明标识了 JWT 的受众，即预期的接收者。
    
  • exp（Expiration Time）：该声明指定了 JWT 的逾期时间，在此时间之后，JWT 将被认为是无效的。
    
  • nbf（Not Before）：该声明指定了 JWT 的生效时间，在此时间之前，JWT 将被认为是无效的。
    
  • iat（Issued At）：该声明指定了 JWT 的签发时间。
    
  • jti（JWT ID）：该声明为 JWT 提供了一个唯一标识符。
    
  
  
• signature：为了确保 JWT 未被篡改，必要对编码后的 header 和 payload 使用指定的签名算法和一个密钥进行签名。
  
  
  
  

代码

1. def jwt_auth(f):
2.     @wraps(f)
3.     def decorated_function(*args, **kwargs):
4.         resp = {
5.             "requestid": uuid.uuid4()
6.         }
7.         auth_header = request.headers.get('Authorization')
8.         if not auth_header or not auth_header.startswith("Bearer "):
9.             resp["message"] = "no Authorization header or invalid format"
10.             return jsonify(resp), 400
11.         try:
12.             token = auth_header.split(' ')[1]
13.             jwt.decode(token, JWT_SECRET, algorithms=['HS256'])
14.         except jwt.ExpiredSignatureError:
15.             resp["error"] = "token has expired"
16.             log.logger.error(f"url:{request.url},token:{token},resp:{resp}")
17.             return jsonify(resp), 401
18.         except jwt.InvalidTokenError:
19.             resp["message"] = f"invalid token {token}"
20.             log.logger.error(f"url:{request.url},token:{token},resp:{resp}")
21.             return jsonify(resp), 401
22.         except Exception as e:
23.             resp["message"] = f"interal error {e}"
24.             log.logger.error(f"url:{request.url},token:{token},resp:{resp}")
25.             return jsonify(resp), 500
26.         return f(*args, **kwargs)
28.     return decorated_function
30. # 用户登录获取jwt token接口
31. @app.route('/api/v2.0/admin/jwt/login', methods=['POST'])
32. def get_jwt_token():
33.     resp = {
34.         "requestid": uuid.uuid4()
35.     }
36.     if request.is_json:
37.         data = request.get_json()
38.         username = data.get('username')
39.         password = data.get('password')
40.         if username is None or password is None:
41.             resp['error'] = 'username or password is required'
42.             log.logger.error(f"url:{request.url},params:{username},resp:{resp}")
43.             return jsonify(resp),400
45.         if not validate_username(username):
46.             resp['error'] = 'username length should not exceed 10'
47.             log.logger.error(f"url:{request.url},params:{username},resp:{resp}")
48.             return jsonify(resp),400
49.         try:
50.             cli = MysqlCli(MYSQL_HOST, MYSQL_USER, MYSQL_PASSWORD, MYSQL_DATABASE)
51.             sql = f"select * from passwords where username = '{username}' limit 1"
52.             user = cli.select_one(sql)
53.             if not user:
54.                 resp['error'] = 'username not found'
55.                 log.logger.error(f"url:{request.url},params:{username},resp:{resp}")
56.                 return jsonify(resp), 400
57.             if user['password'] != password:
58.                 resp['error'] = 'password is not right'
59.                 log.logger.error(f"url:{request.url},params:{username},resp:{resp}")
60.                 return jsonify(resp), 401
61.             payload = {
62.                 "iss": "lady_killer9",
63.                 "exp": datetime.now() + timedelta(seconds=5*60),
64.                 "jti": str(uuid.uuid4())
65.             }
66.             resp['message'] = f'success to login :{user},token:{jwt.encode(payload,JWT_SECRET,algorithm="HS256")}'
67.             log.logger.info(f"url:{request.url},params:{username},resp:{resp}")
68.             return jsonify(resp), 200
69.         except Exception:
70.             resp['error'] = f'failed to get user, error:{traceback.format_exc()}'
71.             log.logger.error(f"url:{request.url},params:{username},resp:{resp}")
72.             return jsonify(resp),500
73.     else:
74.         resp['error'] = "Invalid JSON format in request"
75.         return jsonify(resp),400
77. # 管理员注册用户接口 v2.0增加jwt sha256算法
78. @app.route('/api/v2.0/admin/jwt/add_user', methods=['POST'])
79. @jwt_auth
80. def add_user_jwt():
81.     return add_user()
83. # 用户获取信息接口 v2.0 增加jwt sha256算法
84. @app.route('/api/v2.0/jwt/get_user_info', methods=['POST'])
85. @jwt_auth
86. def get_user_info_jwt():
87.     return get_user_info()

复制代码

问题

• 没有调用次数限制，可能被Dos或DDos攻击，造成服务器压力
  
• 没有防止数据篡改或伪造，被抓包后可能被篡改或伪造数据
  
• 没有对重放攻击的防御，抓包后可能重放包
  
• 越权问题，管理员接口任何人都可以调用
  
• SQL注入，没有做参数化查询等防止SQL操纵，可能被拖库
  

优点：

• token可以设置时间限制，逾期后不可调用
  
• payload的aud等参数可以用于鉴权
  
• 可添加自定义payload，方便做其他的功能
  

引入问题：

• jwt token轻易破解
  

bejson jwt在线解密

Hmac

流程

和Digest Auth差不多，可以由客户端生成随机数，这样请求一次即可，随机数不可重复。
主要代码

1. # 验证摘要
2. def check_signature(signture:str,username:str,nonce:int,data:dict):
3.     try:
4.         cli = MysqlCli(MYSQL_HOST, MYSQL_USER, MYSQL_PASSWORD, MYSQL_DATABASE)
5.         sql = f"select * from `secrets` where `username` = '{username}' limit 1"
6.         secret = cli.select_one(sql)
7.         cli.close()
8.         if secret:
9.             server_signature = hmac.new(str(secret['secret']).encode('utf-8'), json.dumps(data).encode('utf-8'), hashlib.sha256).hexdigest()
10.             log.logger.info(f"secret:{str(secret['secret'])},data:{data},server_signature:{server_signature}")
11.             return  server_signature == signture
12.         log.logger.error(f"check_signature failed,error:{username} {secret}")
13.         return False
14.     except Exception:
15.         log.logger.error(f"check_signature failed,error:{traceback.format_exc()}")
16.         return False
18. def hmac_auth(f):
19.     @wraps(f)
20.     def decorated_function(*args, **kwargs):
21.         resp = {
22.             "requestid": uuid.uuid4()
23.         }
24.         # query检查 nonce username
25.         nonce = request.args.get('nonce', type=int)
26.         username = request.args.get('username')
27.         if nonce is None or username is None:
28.             resp['error'] = 'username or nonce not found in query string'
29.             log.logger.error(f"url:{request.url},resp:{resp}")
30.             return jsonify(resp)
31.         # 随机数验证
32.         if check_nonce(nonce):
33.             log.logger.error(f"check_signature failed,error:{nonce} is in database")
34.             resp['error'] = f"check_signature failed,error:{nonce} is in database"
35.             log.logger.error(f"url:{request.url},resp:{resp}")
36.             return jsonify(resp)
37.         save_nonce_with_expiry(nonce,1,MIN*60)
38.         # signature检查
39.         if "Signature" not in request.headers:
40.             resp['error'] = 'signature is required in headers'
41.             log.logger.error(f"url:{request.url},resp:{resp}")
42.             return jsonify(resp)
43.         # request是json
44.         if request.is_json:
45.             data = request.get_json()
46.             if not check_signature(request.headers['Signature'], username, nonce, data):
47.                 resp['error'] = f'signature {request.headers["Signature"]} is invalid'
48.                 log.logger.error(f"url:{request.url},resp:{resp}")
49.                 return jsonify(resp),401
50.         else:
51.             resp['error'] = "Invalid JSON format in request"
52.             return jsonify(resp),400
53.         return f(*args, **kwargs)
55.     return decorated_function
56. # 管理员注册用户接口 v2.0增加hmac sha256算法
57. @app.route('/api/v2.0/admin/hmac/add_user', methods=['POST'])
58. @hmac_auth
59. def add_user_hmac():
60.     resp = {
61.         "requestid": uuid.uuid4()
62.     }
63.     # request是json
64.     if request.is_json:
65.         data = request.get_json()
66.         username = data.get('username')
67.         if username is None:
68.             resp['error'] = 'username is required'
69.             log.logger.error(f"url:{request.url},params:{username},resp:{resp}")
70.             return jsonify(resp)
71.         phone_number = data.get('phone_number')
72.         if phone_number is None:
73.             resp['error'] = 'phone_number is required'
74.             log.logger.error(f"url:{request.url},params:{phone_number},resp:{resp}")
75.             return jsonify(resp)
77.         if not validate_username(username):
78.             resp['error'] = 'username length should not exceed 10'
79.             log.logger.error(f"url:{request.url},params:{username},resp:{resp}")
80.             return jsonify(resp)
81.         try:
82.             cli = MysqlCli(MYSQL_HOST,MYSQL_USER,MYSQL_PASSWORD,MYSQL_DATABASE)
83.             cli.insert_one("users",
84.                 {
85.                     "username": username,
86.                     "phone_number": phone_number
87.                 }
88.             )
89.             secret = secrets.token_hex(16)
90.             cli.insert_one("secrets",
91.                            {
92.                                "username": username,
93.                                "secret": secret
94.                            })
95.             cli.close()
96.             resp['message'] =f'success to add user:{username},secret {secret},remember it!'
97.             log.logger.info(f"url:{request.url},params:{username},resp:{resp}")
98.         except Exception:
99.             log.logger.info(f"url:{request.url},params:{username},resp:{resp}")
100.             resp['error'] = f"failed to insert to mysql:{traceback.format_exc()}"
101.         return jsonify(resp)
102.     else:
103.         resp['error'] = "Invalid JSON format in request"
104.         return jsonify(resp)
106. # 用户获取信息接口 v2.0 增加hmac
107. @app.route('/api/v2.0/hmac/get_user_info', methods=['POST'])
108. @hmac_auth
109. def get_user_info_hmac():
110.     return get_user_info()

复制代码

问题

• 没有调用次数限制，可能被Dos或DDos攻击，造成服务器压力
  
• 没有防止数据篡改或伪造，被抓包后可能被篡改或伪造数据
  
• 没有对重放攻击的防御，抓包后可能重放包
  
• 越权问题，管理员接口任何人都可以调用
  
• SQL注入，没有做参数化查询等防止SQL操纵，可能被拖库
  

引入问题：

• 必要存储nonce，这里有设置逾期时间
  

OAuth

内容太多，留坑先不看了
比较

比较项Basic AuthAPI keyBearer Auth/Token authDigest AuthHmac AuthJWT身份认证√√√√√√密钥加密×××√√×服务端存储√√√√√×（token可以不存）防重放×××√√×（token失效前可重放）时效性×××√√√自定义××√√√√ 通过以上比较，假如设计一个签名具有以下优点会比较好：

• 防重放。通过随机数防止请求重放，随机数由客户端盘算可以减轻服务端压力。
  
• 密钥加密。使用密钥盘算签名。
  
• 服务端存储。存储随机数，设置逾期时间。
  
• 时效性。添加时间戳，逾期后请求失败。
  
• 自定义。自定义请求签名当然可以自定义一些东西，用于鉴权等。
  

自定义请求签名

身份认证&密钥加密

请求必要验证身份，就必要有账密，这里就用SecretId、SecretKey，其中SecretKey用于进行签名的盘算。
防重放

为了防重放，生成一个随机数Nonce，Nonce唯一，这样服务端收到携带该Nonce的请求后，还发送带该Nonce的请求，就拒绝掉。因此Nonce必要服务器保存，同时为了防止篡改，签名时必要。
那么问题来了，Nonce必要服务器保存，不能不停保存吧，随着时间推移，存储成本会越来越高，因此必要时间限制。
请求时效性

请求应该具有时效性，这里使用unix时间戳Timestamp。规定在1分钟内请求有效，这样Nonce保存时间在1分钟即可。
请求签名算法设计

密钥加密选择SHA-256算法，当然算法可以当做参数，由客户端指定，就用Algorithm吧
url类似：xxx?SecretId=xxx&Nonce=xxx&Timestamp=xxx&Algorithm=xxx
添加一个自定义请求头Signature，放上签名，待签名字符串规定格式如下
   {Nonce}:{Timestamp}:{Algorithm}:{HTTPMethod}:{base64(HTTPBody)}
  

• HTTPMethod：HTTP请求方法，例如POST
  
• HTTPBody：HTTP请求体，例如{“name”:“lady_killer9”}
  

当然，还可以添加更多到待签名字符串
客户端

• 生成一个随机字符串（不包罗:）
  
• 对请求头和请求体做base64编码，并按照格式拼接
  
• 使用算法和SecretKey进行签名
  
• 生成时间戳，发送请求
  

服务端

• 从url获取时间戳Timestamp，校验是否在时间内
  
• 从url获取随机字符串Nonce，校验是否在数据库中
  
• 对请求头和请求体做base64编码
  
• 从url获取SecretId后从数据库查询对应的SecretKey
  
• 使用SecretKey和Algorithm算法对拼接同样格式字符串进行签名得到ServerSignature，比较Signature是否与请求头Signature的值同等
  

代码

1. def verify_signature(signature, nonce,timestamp,algorithm,method,body_base64):
2.     format_str = f"{nonce}:{timestamp}:{algorithm}:{method}:{body_base64}"
3.     server_signature = hashlib.sha256(format_str.encode()).hexdigest()
4.     log.logger.info(format_str)
5.     log.logger.info(server_signature)
6.     return signature == server_signature
9. def require_signature(f):
10.     @wraps(f)
11.     def decorated_function(*args, **kwargs):
12.         resp = {
13.             "requestid": uuid.uuid4()
14.         }
15.         secret_id = request.args.get('SecretId', type=str)
16.         nonce = request.args.get('Nonce', type=str)
17.         timestamp = request.args.get('Timestamp', type=int)
18.         algorithm = request.args.get('Algorithm', type=str)
19.         if secret_id is None:
20.             resp['error'] = 'No SecretId in query string'
21.             return jsonify(resp),400
22.         if nonce is None:
23.             resp['error'] = 'No Nonce in query string'
24.             return jsonify(resp),400
25.         if timestamp is None:
26.             resp['error'] = 'No Timestamp in query string'
27.             return jsonify(resp),400
28.         if algorithm is None:
29.             resp['error'] = 'No Algorithm in query string'
30.             return jsonify(resp),400
31.         if algorithm not in ["sha256"]:
32.             resp['error'] = f'can not support {algorithm}'
33.             return jsonify(resp), 400
34.         if (datetime.now() - timedelta(minutes=MIN)).timestamp() > timestamp:
35.             resp['error'] = 'Request is send before 5 mins ago, check Timestamp'
36.             return jsonify(resp), 400
37.         if ':' in nonce:
38.             resp['error'] = 'can not contain : in Nonce, generate a new one'
39.             return jsonify(resp), 400
40.         cli = redis.Redis.from_url(REDIS_URL)
41.         if cli.exists(nonce):
42.             resp['error'] = 'can not request with same Nonce'
43.             return jsonify(resp), 400
44.         else:
45.             cli.setex(nonce, MIN*60, 1)
47.         signature = request.headers.get('Signature')
48.         if not signature:
49.             resp['error'] = 'no Signature in headers'
50.             return jsonify(resp), 400
52.         # 解析 Authorization header，验证签名
53.         body_base64 = bytes.decode(b64encode(json.dumps(request.get_json(),ensure_ascii=False).encode()))
54.         if not verify_signature(signature, nonce,timestamp,algorithm,request.method,body_base64):
55.             resp['error'] = 'invalid signature'
56.             return jsonify(resp), 401
58.         return f(*args, **kwargs)
60.     return decorated_function

复制代码

攻击与防御

SQL注入

例如，在v1.0的get_user_info接口，存在将用户输入拼接到sql的毛病，可以被SQL注入。

抓包如下：

防御方面可以通过预编译等方式来解决
v3.0已解决

敏感信息泄露

例如，在v1.0的get_user_info接口，用户手机号被完整返回，没有打码。

防御上可以通过加*打码或MFA等来解决
v3.0已解决

越权攻击

例如，在v2.0的api key认证接口，任意用户都能查询admin用户的信息，只必要知道username即可

防御方面可以通过添加RBAC等鉴权来解决
重放攻击

例如，v2.0的api key认证接口，设置burpsuite代理，放到重放器Reapter，发送多少次都可以。

v3.0通过自定义请求签名就解决了此类问题。

全部代码

Github-api_history
参考

API-Security Owasp top 10
API 鉴权都有哪些分类，这些重点不要错过
best-practices-for-authentication-and-authorization-for-rest-apis/
pyjwt
https://github.com/ticarpi/jwt_tool
rfc6750-The OAuth 2.0 Authorization Framework: Bearer Token Usage
rfc7616-HTTP Digest Access Authentication
rfc2617-HTTP Authentication: Basic and Digest Access Authentication
rfc7519-JSON web Token (JWT)
Github-jwt_tool

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。

---

欢迎光临 ToB企服应用市场:ToB评测及商务社交产业平台 (https://dis.qidao123.com/)

Powered by Discuz! X3.4