设置EndPoint和凭证
移动终端是一个不受信任的环境，把AccessKeyId和AccessKeySecret直接保存在终端用来加签请求，存在极高的风险。建议只在测试时使用明文设置模式，业务应用推荐使用STS鉴权模式或自签名模式，详细请参考：访问控制、移动端直传。
如果用STS鉴权模式，推荐使用OSSAuthCredentialProvider方式直接访问鉴权应用服务器，token过期后可以自动更新。
更多信息可查看可查看sample 点击查看。
设置EndPoint和CredentialProvider示例如下：
NSString *endpoint = "http://oss-cn-hangzhou.aliyuncs.com";//直接访问鉴权服务器（推荐，token过期后可以自动更新）id<OSSCredentialProvider> credential = [[OSSAuthCredentialProvider alloc] initWithAuthServerUrl:@"鉴权服务器地址，例如http://abc.com"];client = [[OSSClient alloc] initWithEndpoint:endPoint credentialProvider:credential];
提示：
- 更多鉴权方式参考：访问控制。
初始化_iOS-SDK_SDK 参考_对象存储 OSS-阿里云 https://help.aliyun.com/document_detail/32057.html
访问控制_iOS-SDK_SDK 参考_对象存储 OSS-阿里云 https://help.aliyun.com/document_detail/32059.html
使用STS详细步骤
-
App用户登录。
App用户由您自己管理。您可以自定义身份管理系统，也可以使用外部Web账号或OpenID。对于每个有效的App用户来说，AppServer是可以确切地定义出每个App用户的最小访问权限。
-
AppServer请求STS服务获取一个安全令牌(SecurityToken)。
在调用STS之前，AppServer需要确定App用户的最小访问权限（用Policy语法描述）以及授权的过期时间。
通过调用STS的AssumeRole(扮演角色)接口来获取安全令牌。角色管理与使用相关内容请参考RAM使用指南中的角色管理。
STS返回给AppServer一个有效的访问凭证，包括一个安全令牌(SecurityToken)、临时访问密钥(AccessKeyId, AccessKeySecret)以及过期时间。
-
AppServer将访问凭证返回给ClientApp。
ClientApp可以缓存这个凭证。当凭证失效时，ClientApp需要向AppServer申请新的有效访问凭证。比如，访问凭证有效期为1小时，那么ClientApp可以每30分钟向AppServer请求更新访问凭证。
ClientApp使用本地缓存的访问凭证去请求Aliyun Service API。云服务会感知STS访问凭证，并会依赖STS服务来验证访问凭证，并正确响应用户请求。
注意：使用这种模式授权需要先开通阿里云RAM服务：单击查看
STS安全令牌详情请参考RAM使用指南中的角色管理
调用STS服务接口AssumeRole来获取有效访问凭证即可
直接使用STS SDK来调用该方法单击查看
STS使用手册：单击查看
OSS授权策略配置：单击查看
实战指南：单击查看
import json
from flask import Flask, jsonify, abort, make_response, request
import time
from aliyunsdkcore import client
from aliyunsdksts.request.v20150401 import AssumeRoleRequest def getSts(uid):
# 通过管理控制后台-访问控制 https://help.aliyun.com/product/28625.html
# RAM控制台 https://ram.console.aliyun.com/
# STS授权相关信息获取步骤:
# 1.RAM控制台用户管理创建子用户(User)同时点击该用户创建并获取AccessKeyID和AccessKeySecret https://help.aliyun.com/document_detail/28637.html
# 2.对该子用户(User) 授予AliyunSTSAssumeRoleAccess策略(必须),如需自定义策略请看 https://help.aliyun.com/document_detail/28640.html
# 3.RAM控制台角色管理创建角色role,进行自定义授权设置(控制操作的内容),获取Arn https://help.aliyun.com/document_detail/28649.html
# 注意点：
# 只有子用户(User)才能调用 AssumeRole 接口
# 阿里云主用户(Root User)的AccessKeys不能用于发起AssumeRole请求
# python sdk说明
# 构建一个 Aliyun Client, 用于发起请求
# 构建Aliyun Client时需要设置AccessKeyId和AccessKeySevcret
# STS是Global Service, API入口位于华东 1 (杭州) , 这里Region填写"cn-hangzhou"
# clt = client.AcsClient('<access-key-id>','<access-key-secret>','cn-hangzhou')
AccessKeyID = "************************"
AccessKeySecret = "************************"
roleArn = "************************" '''
root
accessKeyId = '1'
accessKeySecret = '2'
''' '''
//AccessKey详情
AccessKeyID：
1
AccessKeySecret：
2
'''
kid, ks = '1', '2'
AccessKeyID, AccessKeySecret = kid, ks
roleArn = 'acs:ram::3:role/aliyunosstokengeneratorrole'
clt = client.AcsClient(AccessKeyID, AccessKeySecret, 'cn-hangzhou')
# 构造"AssumeRole"请求
request___ = AssumeRoleRequest.AssumeRoleRequest()
# 指定角色 需要在 RAM 控制台上获取
request___.set_RoleArn(roleArn)
# RoleSessionName 是临时Token的会话名称，自己指定用于标识你的用户，主要用于审计，或者用于区分Token颁发给谁
# 但是注意RoleSessionName的长度和规则，不要有空格，只能有'-' '.' '@' 字母和数字等字符
# 具体规则请参考API文档中的格式要求 '''
#AssumeRole_操作接口_API 参考（STS）_访问控制-阿里云 https://help.aliyun.com/document_detail/28763.html RoleSessionName 类型：String
必须：是
描述：用户自定义参数。此参数用来区分不同的Token，可用于用户级别的访问审计。
格式：^[a-zA-Z0-9\.@\-_]+$ 2-32个字符
'''
request___.set_RoleSessionName(uid) # OSS Policy settings could not set by default
# can read https://help.aliyun.com/document_detail/56288.html
# case https://help.aliyun.com/knowledge_detail/39717.html?spm=5176.product28625.6.735.5etPTf
# case https://help.aliyun.com/knowledge_detail/39712.html?spm=5176.7739717.6.729.aZiRgD
# 发起请求，并得到response
try:
response = clt.do_action_with_exception(request___)
## { "ErrorDump": "the JSON object must be str, not 'bytes'", "StatusCode": "500" }
# text = json.loads(response) win ok linux + .decode('utf-8') 加后 win 依然ok
text = json.loads(response.decode('utf-8')) stsDict = dict().fromkeys(
['RequestId', 'uid', 'Expiration', 'AccessKeyId', 'AccessKeySecret', 'SecurityToken', 'StatusCode'])
stsDict["RequestId"] = text["RequestId"]
stsDict["uid"] = uid
stsDict['Expiration'] = text["Credentials"]['Expiration']
stsDict["AccessKeyId"] = text["Credentials"]["AccessKeyId"]
stsDict["AccessKeySecret"] = text["Credentials"]["AccessKeySecret"]
stsDict['SecurityToken'] = text["Credentials"]['SecurityToken']
stsDict["StatusCode"] = "200" # stsText = json.dumps(stsDict)
print('-----------》')
print(stsDict)
print('《-----------')
return stsDict
except Exception as e:
print(e)
# errorDict = dict().fromkeys(['StatusCode', 'ErrorCode', 'ErrorMessage'])
errorDict = dict().fromkeys(['StatusCode', 'ErrorDump'])
errorDict["StatusCode"] = "500"
# errorDict["ErrorMessage"] = e.message
# errorDict["ErrorCode"] = e.error_code
errorDict["ErrorDump"] = '{}'.format(e)
# stsText = json.dumps(errorDict)
return errorDict
# return stsText
pass aliBridge = Flask(__name__)
aliBridge.config['JSON_AS_ASCII'] = False @aliBridge.route('/v1.0/aliBridge/aliyunosstokengenerator', methods=['POST'])
def aliyunosstokengeneratorrole():
if not request.json:
abort(400)
chk_k = ['ipv4', 'imei', 'kid', 'ks']
ipv4, imei, kid, ks = request.json['ipv4'], request.json['imei'], request.json['kid'], request.json['ks']
# ipv4, imei, kid, ks = request.json[0:4]
uid = 'ipv4@{}@imei@{}'.format(ipv4, imei)
uid = '{}@{}'.format(ipv4, imei)
# uid ='ipv4__123__imei_123',
if (kid, ks) != ('ourServerKeyId', 'ourServerKeyValue'):
abort(400)
for c in chk_k:
if c not in request.json:
abort(400)
# task = {
# 'uid': request.json['uid'],
# # 'uid': request,
# 'no_open': 12,
# 'no_ad': 34,
# 'description': '该uid在ad_direct_order*计123条当前未失效，其中12条打不开相应页面，34条页面中没有我司广告位',
# 'cost_time': 123,
# 'request_time': time.strftime('%Y%m%d_%H%M%S', time.localtime(time.time() - 123)),
# 'current_time': time.strftime('%Y%m%d_%H%M%S', time.localtime(time.time()))
# }
# return jsonify({'status': '1', 'info': task}), 201 task = getSts(uid) return jsonify(task), 201 '''
{
"ipv4":"197.164.165.154","imei":"123456789012347","kid":"ourServerKeyId","ks":"ourServerKeyValue"
}
''' if __name__ == '__main__':
aliBridge.run(host='0.0.0.0', port=5001, debug=True)