网络安全-如何设计一个安全的API（安全角度）

API安全概述

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

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

有以下常见的防御方式：

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

当然，有些还是需要API后端代码来进行防御，例如命令注入、SSRF等。

本文以API身份认证为主要内容，浅谈各种认证的使用场景与优缺点，同时穿插部分攻击与防御。

设计一个安全的API

一个基本的API

主要代码

import uuid
import re
import traceback

from flask import Flask, request, jsonify
from mysql import MysqlCli
from log import log
from setting import*

app = Flask(__name__)
log.set_file()# 验证usernamedefvalidate_username(username:str)->bool:iflen(username)>20:returnFalsereturnTrue# 验证手机号defvalidate_phone_number(phone_number:str)->bool:# 使用正则表达式检查手机号格式
    pattern = re.compile(r'^1[3456789]\d{9}$')if re.match(pattern, phone_number):returnTrueelse:returnFalse# 管理员注册用户接口@app.route('/api/v1.0/admin/add_user', methods=['POST'])defadd_user():
    resp ={"requestid": uuid.uuid4()}if request.is_json:
        data = request.get_json()
        username = data.get('username')if username isNone:
            resp['error']='username is required'
            log.logger.error(f"url:{request.url},params:{username},resp:{resp}")return jsonify(resp),400
        phone_number = data.get('phone_number')if phone_number isNone:
            resp['error']='phone_number is required'
            log.logger.error(f"url:{request.url},params:{phone_number},resp:{resp}")return jsonify(resp),400ifnot validate_username(username):
            resp['error']='username length should not exceed 10'
            log.logger.error(f"url:{request.url},params:{username},resp:{resp}")return jsonify(resp),400try:
            cli = MysqlCli(MYSQL_HOST,MYSQL_USER,MYSQL_PASSWORD,MYSQL_DATABASE)
            cli.insert_one("users",{"username": username,"phone_number": phone_number
                })
            cli.close()
            resp['message']=f'success to add user:{username}!'
            log.logger.info(f"url:{request.url},params:{username},resp:{resp}")return jsonify(resp)except Exception:
            log.logger.info(f"url:{request.url},params:{username},resp:{resp}")
            resp['error']=f"failed to insert to mysql:{traceback.format_exc()}"return jsonify(resp),500else:
        resp['error']="Invalid JSON format in request"return jsonify(resp),400# 用户获取信息接口@app.route('/api/v1.0/get_user_info', methods=['POST'])defget_user_info():
    resp ={"requestid": uuid.uuid4()}if request.is_json:
        data = request.get_json()
        username = data.get('username')if username isNone:
            resp['error']='username is required'
            log.logger.error(f"url:{request.url},params:{username},resp:{resp}")return jsonify(resp),400ifnot validate_username(username):
            resp['error']='username length should not exceed 10'
            log.logger.error(f"url:{request.url},params:{username},resp:{resp}")return jsonify(resp),400try:
            cli = MysqlCli(MYSQL_HOST, MYSQL_USER, MYSQL_PASSWORD, MYSQL_DATABASE)
            sql =f"select * from users where username = '{username}' limit 1"
            user = cli.select_all(sql)
            resp['message']=f'success to get user:{user}.'
            log.logger.info(f"url:{request.url},params:{username},resp:{resp}")return jsonify(resp)except Exception:
            resp['error']=f'failed to get user, error:{traceback.format_exc()}'
            log.logger.error(f"url:{request.url},params:{username},resp:{resp}")return jsonify(resp),500else:
        resp['error']="Invalid JSON format in request"return jsonify(resp),400if __name__ =='__main__':
    app.run()

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

版本控制
日志记录
请求方法校验
请求数据校验

调用

在这里插入图片描述

API的一些问题

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

BasicAuth

认证流程

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

主要代码

defbasic_auth(f):@wraps(f)defdecorated_function(*args,**kwargs):
        resp ={"requestid": uuid.uuid4()}
        auth_header = request.headers.get('Authorization')ifnot auth_header ornot auth_header.startswith('Basic '):
            resp['error']='basic auth is required'
            log.logger.error(f"url:{request.url},resp:{resp}")return jsonify(resp),400else:try:
                encoded_credentials = auth_header.split(' ')[1]
                decoded_credentials = base64.b64decode(encoded_credentials).decode('utf-8')
                username, password = decoded_credentials.split(':')ifnot check_basic_auth(username, password):
                    resp['error']='basic auth failed,check your username or password is right'
                    log.logger.error(f"url:{request.url},resp:{resp}")return jsonify(resp),401except Exception:
                resp['error']=f'basic auth failed,err: {traceback.format_exc()}'
                log.logger.error(f"url:{request.url},resp:{resp}")return jsonify(resp),500return f(*args,**kwargs)return decorated_function
# 管理员注册用户接口 v2.0 增加密码@app.route('/api/v2.0/admin/basic_auth/add_user', methods=['POST'])@basic_authdefadd_user_basic_auth():
    resp ={"requestid": uuid.uuid4()}# 请求是json格式if request.is_json:
        data = request.get_json()
        username = data.get('username')# username检查if username isNone:
            resp['error']='username is required'
            log.logger.error(f"url:{request.url},params:{username},resp:{resp}")return jsonify(resp),400ifnot validate_username(username):
            resp['error']='username length should not exceed 10'
            log.logger.error(f"url:{request.url},params:{username},resp:{resp}")return jsonify(resp),400# phone_number检查
        phone_number = data.get('phone_number')if phone_number isNone:
            resp['error']='phone_number is required'
            log.logger.error(f"url:{request.url},params:{phone_number},resp:{resp}")return jsonify(resp),400# 插入数据库try:
            cli = MysqlCli(MYSQL_HOST,MYSQL_USER,MYSQL_PASSWORD,MYSQL_DATABASE)
            cli.insert_one("users",{"username": username,"phone_number": phone_number
                })
            pwd = generate_random_password(secrets.choice(range(8,17)))
            cli.insert_one("passwords",{"username": username,"password": pwd
            })
            cli.close()
            resp['message']=f'success to add user:{username},password {pwd},remember it!'
            log.logger.info(f"url:{request.url},params:{username},resp:{resp}")# 异常返回except Exception:
            log.logger.info(f"url:{request.url},params:{username},resp:{resp}")
            resp['error']=f"failed to insert to mysql:{traceback.format_exc()}",500return jsonify(resp)return jsonify(resp)# 请求不是json格式else:
        resp['error']="Invalid JSON format in request"return jsonify(resp),400

在这里插入图片描述

问题

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

引入问题：

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

API Key

流程

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

主要代码

defapi_key_auth(f):@wraps(f)defdecorated_function(*args,**kwargs):
        resp ={"requestid": uuid.uuid4()}# api key检查if"api_key"notin request.headers:
            resp['error']='api_key is required in headers'
            log.logger.error(f"url:{request.url},resp:{resp}")return jsonify(resp),400ifnot check_api_key(request.headers["api_key"]):
            resp['error']=f'api_key {request.headers["api_key"]} is invalid'
            log.logger.error(f"url:{request.url},resp:{resp}")return jsonify(resp),401return f(*args,**kwargs)return decorated_function
# 管理员注册用户接口 v2.0 增加api [email protected]('/api/v2.0/admin/api_key/add_user', methods=['POST'])@api_key_authdefadd_user_api_key():
    resp ={"requestid": uuid.uuid4()}# request是jsonif request.is_json:
        data = request.get_json()
        username = data.get('username')if username isNone:
            resp['error']='username is required'
            log.logger.error(f"url:{request.url},params:{username},resp:{resp}")return jsonify(resp),400
        phone_number = data.get('phone_number')if phone_number isNone:
            resp['error']='phone_number is required'
            log.logger.error(f"url:{request.url},params:{phone_number},resp:{resp}")return jsonify(resp),400ifnot validate_username(username):
            resp['error']='username length should not exceed 10'
            log.logger.error(f"url:{request.url},params:{username},resp:{resp}")return jsonify(resp),400try:
            cli = MysqlCli(MYSQL_HOST,MYSQL_USER,MYSQL_PASSWORD,MYSQL_DATABASE)
            cli.insert_one("users",{"username": username,"phone_number": phone_number
                })
            key = secrets.token_hex(16)
            cli.insert_one("keys",{"username": username,"key": key
                           })
            cli.close()
            resp['message']=f'success to add user:{username},key {key},remember it!'
            log.logger.info(f"url:{request.url},params:{username},resp:{resp}")except Exception:
            log.logger.info(f"url:{request.url},params:{username},resp:{resp}")
            resp['error']=f"failed to insert to mysql:{traceback.format_exc()}"return jsonify(resp),500return jsonify(resp)else:
        resp['error']="Invalid JSON format in request"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哈希。

主要代码

# 将字符串保存到Redis中，并设置过期时间defsave_nonce_with_expiry(key, value, expiry_seconds):"""
    :param key: 键
    :param value: 值
    :param expiry_seconds: 过期时间（秒）
    """# 连接Redis数据库
    redis_client = redis.StrictRedis(host=REDIS_HOST,password=REDIS_PASSWORD, port=REDIS_PORT, db=REDIS_DB)
    redis_client.setex(key, expiry_seconds, value)defcheck_nonce(key):"""
    检查字符串是否存在于Redis中
    :param key: 键
    :return: 布尔值，表示键是否存在
    """# 连接Redis数据库
    redis_client = redis.StrictRedis(host=REDIS_HOST,password=REDIS_PASSWORD ,port=REDIS_PORT, db=REDIS_DB)return redis_client.exists(key)# 校验username 返回密码defcheck_username(username):try:
        cli = MysqlCli(MYSQL_HOST, MYSQL_USER, MYSQL_PASSWORD, MYSQL_DATABASE)
        sql =f"select * from passwords where username = '{username}' limit 1"
        user = cli.select_one(sql)return user['password']if user else''except Exception:
        log.logger.error(f"check_basic_auth failed,error:{traceback.format_exc()}")returnFalse# 校验responsedefcheck_response(response,realm,username,password,method,uri,req_nonce,nc,cnonce,qop):try:
        log.logger.info(f"check_response ,response:{response},realm:{realm},username:{username},"f"password:{password},method:{method},uri:{uri},req_nonce:{req_nonce},nc:{nc},"f"cnonce:{cnonce},qop:{qop}")# 校验nonceifnot check_nonce(req_nonce):
            log.logger.error(f"check_signature failed,error:{req_nonce} not exist!")returnFalse
        ha1=hashlib.md5(f"{username}:{realm}:{password}".encode()).hexdigest()
        ha2=hashlib.md5(f"{method}:{uri}".encode()).hexdigest()return response == hashlib.md5(f"{ha1}:{req_nonce}:{nc}:{cnonce}:{qop}:{ha2}".encode()).hexdigest()except Exception:
        log.logger.error(f"check_responce failed,error:{traceback.format_exc()}")returnFalse# 添加用户 用于添加用户时无需额外增加用户关联的密钥等信息时的接口defadd_user():
    resp ={"requestid": uuid.uuid4()}
    log.logger.info(request.headers)# request是jsonif request.is_json:
        data = request.get_json()
        username = data.get('username')if username isNone:
            resp['error']='username is required'
            log.logger.error(f"url:{request.url},params:{username},resp:{resp}")return jsonify(resp)
        phone_number = data.get('phone_number')if phone_number isNone:
            resp['error']='phone_number is required'
            log.logger.error(f"url:{request.url},params:{phone_number},resp:{resp}")return jsonify(resp)ifnot validate_username(username):
            resp['error']='username length should not exceed 10'
            log.logger.error(f"url:{request.url},params:{username},resp:{resp}")return jsonify(resp)try:
            cli = MysqlCli(MYSQL_HOST, MYSQL_USER, MYSQL_PASSWORD, MYSQL_DATABASE)
            cli.insert_one("users",{"username": username,"phone_number": phone_number
                           })
            cli.close()
            resp['message']=f'success to add user:{username}!'
            log.logger.info(f"url:{request.url},params:{username},resp:{resp}")except Exception:
            log.logger.info(f"url:{request.url},params:{username},resp:{resp}")
            resp['error']=f"failed to insert to mysql:{traceback.format_exc()}"return jsonify(resp)else:
        resp['error']="Invalid JSON format in request"return jsonify(resp)# 管理员注册用户接口 v2.0增加digest算法@app.route('/api/v2.0/admin/digest_auth/add_user', methods=['POST'])@digest_authdefadd_user_digest():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 使用指定的签名算法和一个密钥进行签名。

代码

defjwt_auth(f):@wraps(f)defdecorated_function(*args,**kwargs):
        resp ={"requestid": uuid.uuid4()}
        auth_header = request.headers.get('Authorization')ifnot auth_header ornot auth_header.startswith("Bearer "):
            resp["message"]="no Authorization header or invalid format"return jsonify(resp),400try:
            token = auth_header.split(' ')[1]
            jwt.decode(token, JWT_SECRET, algorithms=['HS256'])except jwt.ExpiredSignatureError:
            resp["error"]="token has expired"
            log.logger.error(f"url:{request.url},token:{token},resp:{resp}")return jsonify(resp),401except jwt.InvalidTokenError:
            resp["message"]=f"invalid token {token}"
            log.logger.error(f"url:{request.url},token:{token},resp:{resp}")return jsonify(resp),401except Exception as e:
            resp["message"]=f"interal error {e}"
            log.logger.error(f"url:{request.url},token:{token},resp:{resp}")return jsonify(resp),500return f(*args,**kwargs)return decorated_function

# 用户登录获取jwt token接口@app.route('/api/v2.0/admin/jwt/login', methods=['POST'])defget_jwt_token():
    resp ={"requestid": uuid.uuid4()}if request.is_json:
        data = request.get_json()
        username = data.get('username')
        password = data.get('password')if username isNoneor password isNone:
            resp['error']='username or password is required'
            log.logger.error(f"url:{request.url},params:{username},resp:{resp}")return jsonify(resp),400ifnot validate_username(username):
            resp['error']='username length should not exceed 10'
            log.logger.error(f"url:{request.url},params:{username},resp:{resp}")return jsonify(resp),400try:
            cli = MysqlCli(MYSQL_HOST, MYSQL_USER, MYSQL_PASSWORD, MYSQL_DATABASE)
            sql =f"select * from passwords where username = '{username}' limit 1"
            user = cli.select_one(sql)ifnot user:
                resp['error']='username not found'
                log.logger.error(f"url:{request.url},params:{username},resp:{resp}")return jsonify(resp),400if user['password']!= password:
                resp['error']='password is not right'
                log.logger.error(f"url:{request.url},params:{username},resp:{resp}")return jsonify(resp),401
            payload ={"iss":"lady_killer9","exp": datetime.now()+ timedelta(seconds=5*60),"jti":str(uuid.uuid4())}
            resp['message']=f'success to login :{user},token:{jwt.encode(payload,JWT_SECRET,algorithm="HS256")}'
            log.logger.info(f"url:{request.url},params:{username},resp:{resp}")return jsonify(resp),200except Exception:
            resp['error']=f'failed to get user, error:{traceback.format_exc()}'
            log.logger.error(f"url:{request.url},params:{username},resp:{resp}")return jsonify(resp),500else:
        resp['error']="Invalid JSON format in request"return jsonify(resp),400# 管理员注册用户接口 v2.0增加jwt sha256算法@app.route('/api/v2.0/admin/jwt/add_user', methods=['POST'])@jwt_authdefadd_user_jwt():return add_user()# 用户获取信息接口 v2.0 增加jwt sha256算法@app.route('/api/v2.0/jwt/get_user_info', methods=['POST'])@jwt_authdefget_user_info_jwt():return get_user_info()

在这里插入图片描述

问题

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

优点：

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

引入问题：

jwt token容易破解

bejson jwt在线解密在这里插入图片描述

Hmac

流程

在这里插入图片描述
和Digest Auth差不多，可以由客户端生成随机数，这样请求一次即可，随机数不可重复。

主要代码

# 验证摘要defcheck_signature(signture:str,username:str,nonce:int,data:dict):try:
        cli = MysqlCli(MYSQL_HOST, MYSQL_USER, MYSQL_PASSWORD, MYSQL_DATABASE)
        sql =f"select * from `secrets` where `username` = '{username}' limit 1"
        secret = cli.select_one(sql)
        cli.close()if secret:
            server_signature = hmac.new(str(secret['secret']).encode('utf-8'), json.dumps(data).encode('utf-8'), hashlib.sha256).hexdigest()
            log.logger.info(f"secret:{str(secret['secret'])},data:{data},server_signature:{server_signature}")return  server_signature == signture
        log.logger.error(f"check_signature failed,error:{username}{secret}")returnFalseexcept Exception:
        log.logger.error(f"check_signature failed,error:{traceback.format_exc()}")returnFalsedefhmac_auth(f):@wraps(f)defdecorated_function(*args,**kwargs):
        resp ={"requestid": uuid.uuid4()}# query检查 nonce username
        nonce = request.args.get('nonce',type=int)
        username = request.args.get('username')if nonce isNoneor username isNone:
            resp['error']='username or nonce not found in query string'
            log.logger.error(f"url:{request.url},resp:{resp}")return jsonify(resp)# 随机数验证if check_nonce(nonce):
            log.logger.error(f"check_signature failed,error:{nonce} is in database")
            resp['error']=f"check_signature failed,error:{nonce} is in database"
            log.logger.error(f"url:{request.url},resp:{resp}")return jsonify(resp)
        save_nonce_with_expiry(nonce,1,MIN*60)# signature检查if"Signature"notin request.headers:
            resp['error']='signature is required in headers'
            log.logger.error(f"url:{request.url},resp:{resp}")return jsonify(resp)# request是jsonif request.is_json:
            data = request.get_json()ifnot check_signature(request.headers['Signature'], username, nonce, data):
                resp['error']=f'signature {request.headers["Signature"]} is invalid'
                log.logger.error(f"url:{request.url},resp:{resp}")return jsonify(resp),401else:
            resp['error']="Invalid JSON format in request"return jsonify(resp),400return f(*args,**kwargs)return decorated_function
# 管理员注册用户接口 v2.0增加hmac sha256算法@app.route('/api/v2.0/admin/hmac/add_user', methods=['POST'])@hmac_authdefadd_user_hmac():
    resp ={"requestid": uuid.uuid4()}# request是jsonif request.is_json:
        data = request.get_json()
        username = data.get('username')if username isNone:
            resp['error']='username is required'
            log.logger.error(f"url:{request.url},params:{username},resp:{resp}")return jsonify(resp)
        phone_number = data.get('phone_number')if phone_number isNone:
            resp['error']='phone_number is required'
            log.logger.error(f"url:{request.url},params:{phone_number},resp:{resp}")return jsonify(resp)ifnot validate_username(username):
            resp['error']='username length should not exceed 10'
            log.logger.error(f"url:{request.url},params:{username},resp:{resp}")return jsonify(resp)try:
            cli = MysqlCli(MYSQL_HOST,MYSQL_USER,MYSQL_PASSWORD,MYSQL_DATABASE)
            cli.insert_one("users",{"username": username,"phone_number": phone_number
                })
            secret = secrets.token_hex(16)
            cli.insert_one("secrets",{"username": username,"secret": secret
                           })
            cli.close()
            resp['message']=f'success to add user:{username},secret {secret},remember it!'
            log.logger.info(f"url:{request.url},params:{username},resp:{resp}")except Exception:
            log.logger.info(f"url:{request.url},params:{username},resp:{resp}")
            resp['error']=f"failed to insert to mysql:{traceback.format_exc()}"return jsonify(resp)else:
        resp['error']="Invalid JSON format in request"return jsonify(resp)# 用户获取信息接口 v2.0 增加[email protected]('/api/v2.0/hmac/get_user_info', methods=['POST'])@hmac_authdefget_user_info_hmac():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的值一致

代码

defverify_signature(signature, nonce,timestamp,algorithm,method,body_base64):
    format_str =f"{nonce}:{timestamp}:{algorithm}:{method}:{body_base64}"
    server_signature = hashlib.sha256(format_str.encode()).hexdigest()
    log.logger.info(format_str)
    log.logger.info(server_signature)return signature == server_signature

defrequire_signature(f):@wraps(f)defdecorated_function(*args,**kwargs):
        resp ={"requestid": uuid.uuid4()}
        secret_id = request.args.get('SecretId',type=str)
        nonce = request.args.get('Nonce',type=str)
        timestamp = request.args.get('Timestamp',type=int)
        algorithm = request.args.get('Algorithm',type=str)if secret_id isNone:
            resp['error']='No SecretId in query string'return jsonify(resp),400if nonce isNone:
            resp['error']='No Nonce in query string'return jsonify(resp),400if timestamp isNone:
            resp['error']='No Timestamp in query string'return jsonify(resp),400if algorithm isNone:
            resp['error']='No Algorithm in query string'return jsonify(resp),400if algorithm notin["sha256"]:
            resp['error']=f'can not support {algorithm}'return jsonify(resp),400if(datetime.now()- timedelta(minutes=MIN)).timestamp()> timestamp:
            resp['error']='Request is send before 5 mins ago, check Timestamp'return jsonify(resp),400if':'in nonce:
            resp['error']='can not contain : in Nonce, generate a new one'return jsonify(resp),400
        cli = redis.Redis.from_url(REDIS_URL)if cli.exists(nonce):
            resp['error']='can not request with same Nonce'return jsonify(resp),400else:
            cli.setex(nonce, MIN*60,1)

        signature = request.headers.get('Signature')ifnot signature:
            resp['error']='no Signature in headers'return jsonify(resp),400# 解析 Authorization header，验证签名
        body_base64 =bytes.decode(b64encode(json.dumps(request.get_json(),ensure_ascii=False).encode()))ifnot verify_signature(signature, nonce,timestamp,algorithm,request.method,body_base64):
            resp['error']='invalid signature'return jsonify(resp),401return f(*args,**kwargs)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

标签：安全 web安全 api认证

本文转载自: https://blog.csdn.net/lady_killer9/article/details/139220388
版权归原作者 lady_killer9 所有，如有侵权，请联系我们删除。

网络安全-如何设计一个安全的API（安全角度）

目录

API安全概述

设计一个安全的API

一个基本的API

主要代码

调用

API的一些问题

BasicAuth

认证流程

主要代码

问题

API Key

流程

主要代码

问题

Bearer auth/Token auth

流程

Digest Auth

流程

主要代码

问题

JWT Token

流程

代码

问题

Hmac

流程

主要代码

问题

OAuth

比较

自定义请求签名

身份认证&密钥加密

防重放

请求时效性

请求签名算法设计

代码

攻击与防御

SQL注入

敏感信息泄露

越权攻击

重放攻击

全部代码

参考

发表评论

“网络安全-如何设计一个安全的API（安全角度）”的评论:

关于作者

overfit同步小助手

相关阅读

文章导航