> 技术文档 > [肥用云计算] Serverless 调用有鉴权的 HTTP FC_阿里云fc函数 接口

[肥用云计算] Serverless 调用有鉴权的 HTTP FC_阿里云fc函数 接口


Serverless 调用有鉴权的 HTTP FC

前言

在之前的文章中,我们介绍了如何调用 HTTP FC(Serverless),但没有详细说明在有鉴权的情况下该如何进行调用。因此,就有了这篇文章。

本文重点探讨如何调用有鉴权的 HTTP Serverless 函数。在阿里云函数计算中,HTTP 调用有三种认证场景:

  1. 匿名调用:无需认证,任何人都可以访问
  2. 函数鉴权调用:需要函数级别的认证
  3. 自定义鉴权调用:需要自定义的认证机制

本文集中于第二种认证——创建带 HTTP 函数鉴权的 FC。同时将详细介绍两种调用方案:从 FC 调用和从 HTTP Server 调用,并重点说明它们在凭证获取方式上的差异。

技术背景及重点

阿里云函数计算的认证机制

阿里云函数计算提供了完整的认证体系,确保服务间调用的安全性。当函数需要相互调用时,系统会自动分配临时凭证,这些凭证具有时效性,提高了安全性。

函数鉴权调用直接借用了阿里云的认证体系,无需自己实现一套认证机制。

流程如下:

  1. 创建一个带 HTTP 函数鉴权的 FC
  2. 获取阿里云的 AccessKey、AccessKeySecret、Token(FC 和 HTTP Server 都有不同的凭证获取方式,在文章后面会详细说明)
  3. 调用之前创建的 HTTP FC
  4. 阿里云负责鉴权是否有权限

凭证获取方式的差异

这是本文的核心重点:

  • 从 FC 调用:通过 context.credentials 获取临时凭证
  • 从 HTTP Server 调用:通过环境变量获取临时凭证

HTTP FC 调用方式

建议可以和 FC 之间的相互调用的文章一起服用,效果更佳。

下面通过一个简单的例子来说明上述原理。

项目架构设计

项目地址,我们的项目包含三个函数:

  1. 被调用方(answer):提供 HTTP 接口,配置函数鉴权
  2. 调用方(fccaller):使用从 FC 调用,通过 context.credentials 获取凭证,完成调用
  3. 调用方(httpservercaller):使用从 HTTP Server(基于 Flask 的框架)中调用,通过环境变量获取凭证,完成调用

项目结构

├── answer/  # 被调用函数├── fccaller/ # FC 调用方├── httpservercaller/ # HTTP Server 调用方

下面描述各个项目的关键点。

Answer 被调用函数

配置文件

# answer/s.yamlresources: answer: component: fc3 props: ... triggers: - triggerConfig: methods: - GET - POST - PUT - DELETE - OPTIONS authType: function # 关键配置:启用函数鉴权 disableURLInternet: false triggerName: defaultTrigger description: \'\' qualifier: LATEST triggerType: http
  1. 开启了 HTTP trigger
  2. authType: function 添加这个配置后,就开启了函数鉴权

实际部署

# 部署 answer 函数cd answers deploy# 需要获取对应 FC HTTP 接口s info | grep urlInternet# 输出: urlInternet: https://practice-answer-tkkwptdgyr.cn-shenzhen.fcapp.run# 设置环境变量——这个需要根据实际情况换成自己的 URLexport TARGET_FUNCTION_URL=\'https://practice-answer-tkkwptdgyr.cn-shenzhen.fcapp.run\'

从 FC 调用

配置文件

# fccaller/s.yamlrole: acs:ram::1719759326012690:role/fc-invoke-test # 这个是关键,用于标识使用哪个角色进行访问environmentVariables: # 你可以用如下命令设置环境变量(Linux/macOS): # export TARGET_FUNCTION_URL=\'https://practice-answer-tkkwptdgyr.cn-shenzhen.fcapp.run\' TARGET_FUNCTION_URL: ${env(TARGET_FUNCTION_URL)}
  1. 配置了 role
  2. 从环境变量中获取 answer 的 URL,用于访问

调用代码实现片段

fccaller 函数通过 context.credentials 获取临时凭证:

# fccaller/code/index.py # 获取凭证 - 这是关键差异 creds = context.credentials # 设置认证信息 accessKeyId = creds.access_key_id accessKeySecret = creds.access_key_secret securityToken = creds.security_token # 准备请求数据 data = { \"key1\": \"a-val1\", \"key2\": \"a-val2\" } body = json.dumps(data) # 设置目标函数的HTTP触发器URL url = os.environ.get(\'TARGET_FUNCTION_URL\') # 设置请求方法和时间 method = \'POST\' date = datetime.utcnow().isoformat(\'T\')[:19]+\'Z\' # 准备请求头 headers = { \'x-acs-date\': date, \'x-acs-security-token\': securityToken, \'Content-Type\': \'application/json\' } # 解析URL并准备认证请求 parsedUrl = urlparse(url) authRequest = TeaRequest() authRequest.method = method authRequest.pathname = parsedUrl.path.replace(\'$\', \'%24\') authRequest.headers = headers authRequest.query = {k: v[0] for k, v in parse_qs(parsedUrl.query).items()} if parsedUrl.query else {} # 生成认证信息 auth = util.get_authorization(authRequest, \'ACS3-HMAC-SHA256\', \'\', accessKeyId, accessKeySecret) headers[\'authorization\'] = auth # 发送HTTP请求 resp = requests.post(url, body, headers=headers) # 处理响应 print(f\"状态码: {resp.status_code}\") print(f\"原始响应内容: {resp.text}\") try: response_body = resp.json() print(f\"函数调用结果: {response_body}\") except json.JSONDecodeError: response_body = resp.text print(f\"响应不是JSON格式: {response_body}\") return { \'statusCode\': 200, \'body\': json.dumps(response_body) if isinstance(response_body, dict) else response_body } except Exception as e: import traceback print(f\"调用失败: {str(e)}\") print(f\"堆栈跟踪: {traceback.format_exc()}\") return { \'statusCode\': 500, \'body\': f\'调用失败: {str(e)}\' }

实际验证

让我们看看实际的调用过程:

# 设置环境变量——这个需要根据实际情况换成自己的 URLexport TARGET_FUNCTION_URL=\'https://practice-answer-tkkwptdgyr.cn-shenzhen.fcapp.run\'# 部署 fccaller 函数cd ../fccallers deploy# 调用 fccaller 函数s invoke

从日志中可以看到调用结果:

========= FC invoke Logs begin =========FunctionCompute python3 runtime inited.FC Invoke Start RequestId: 1-68999d84-17fdfe2f-8aabb4f547e0Credentials - Access Key ID: XXXXXCredentials - Access Key Secret: XXXXCredentials - Security Token: CAISyAJ1q6Ft5B2yfSjIr5nlANPCg+h11YyBNXz6jGYgT816iJzu2zz2IHhMdHZuAOkatv0wmWtS5vkblqZdVplOWU3Da+B364xK7Q75h2QKLS3uv9I+k5SANTW5KXyShb3/AYjQSNfaZY3eCTTtnTNyxr3XbCirW0ffX7SClZ9gaKZ8PGD6F00kYu1bPQx/ssQXGGLMPPK2SH7Qj3HXEVBjt3gX6wo9y9zmnZTNtUeP1wekkrZP+dWpGPX+MZkwZqUYesyuwel7epDG1CNt8BVQ/M909vccoWeY4IjEXwEIvEjdabqLq8cJIgQ8Z64kHK1J6eTxjuFoBGFSEAhqKn2nVMAisES3LOjIqKOsk2M46lkRum+DJG471MmhNK+IyyAFeSp6VXJ7CysSWA+COu7cOEDpWt9XyhCSIt67EsjcgPldubNTbKm2lbLiGoABHziSFeiAZ7okRMYot/8k7+F3UPVyiiHc42YcdpbQaKqihqo0kE9prAZ+ylPAxPaRv1JhPa+LoP+QqlvM4J2EATuCmPlEcLMgel0y+gf5wiAksS+n3Wu/LD/Moxr0qv4+tC0vYlEJEb59B7MfAoujUaXLrBbpaqiNbjAGAGJedqUgAA==状态码: 200原始响应内容: {\"key1\": \"a-val1\", \"key2\": \"a-val2\", \"from_source\": \"answer\"}函数调用结果: {\'key1\': \'a-val1\', \'key2\': \'a-val2\', \'from_source\': \'answer\'}FC Invoke End RequestId: 1-68999d84-17fdfe2f-8aabb4f547e0Duration: 331.41 ms, Billed Duration: 332 ms, Memory Size: 128 MB, Max Memory Used: 22.98 MB========= FC invoke Logs end =========Invoke Result:{ \"body\": \"{\\\"key1\\\": \\\"a-val1\\\", \\\"key2\\\": \\\"a-val2\\\", \\\"from_source\\\": \\\"answer\\\"}\", \"statusCode\": 200}

可以看到:

  1. 通过 context.credentials 成功获取了临时凭证
  2. 调用成功,状态码为 200
  3. answer 函数返回了带有 from_source: \"answer\" 字段的响应,证明已经调用了之前创建的 answer FC。

从 HTTP Server 请求调用

配置文件

# httpservercaller/s.yamlrole: acs:ram::1719759326012690:role/fc-invoke-test # 这个是关键,用于标识使用哪个角色进行访问environmentVariables: # 你可以用如下命令设置环境变量(Linux/macOS): # export TARGET_FUNCTION_URL=\'https://practice-answer-tkkwptdgyr.cn-shenzhen.fcapp.run\' TARGET_FUNCTION_URL: ${env(TARGET_FUNCTION_URL)}
  1. 配置了 role
  2. 从环境变量中获取 answer 的 URL,用于访问

调用代码实现片段

httpservercaller 函数通过环境变量获取临时凭证:

# httpservercaller/code/index.py@app.route(\"/call\", methods=[\"GET\", \"POST\", \"PUT\", \"DELETE\"])def call(): # 通过环境变量获取凭证 - 这是关键差异 accessKeyId = os.environ.get(\'ALIBABA_CLOUD_ACCESS_KEY_ID\', \'Not set\') accessKeySecret = os.environ.get(\'ALIBABA_CLOUD_ACCESS_KEY_SECRET\', \'Not set\') securityToken = os.environ.get(\'ALIBABA_CLOUD_SECURITY_TOKEN\', \'Not set\') # 打印凭证信息进行调试 print(f\"Credentials - Access Key ID: {accessKeyId}\") print(f\"Credentials - Access Key Secret: {accessKeySecret}\") print(f\"Credentials - Security Token: {securityToken}\") # 准备请求数据 data = { \"key1\": \"http-a-val1\", \"key2\": \"http-a-val2\" } body = json.dumps(data) # 设置目标函数的HTTP触发器URL url = os.environ.get(\'TARGET_FUNCTION_URL\') # 设置请求方法和时间 method = \'POST\' date = datetime.utcnow().isoformat(\'T\')[:19]+\'Z\' # 准备请求头 headers = { \'x-acs-date\': date, \'x-acs-security-token\': securityToken, \'Content-Type\': \'application/json\' } # 解析URL并准备认证请求 parsedUrl = urlparse(url) authRequest = TeaRequest() authRequest.method = method authRequest.pathname = parsedUrl.path.replace(\'$\', \'%24\') authRequest.headers = headers authRequest.query = {k: v[0] for k, v in parse_qs(parsedUrl.query).items()} if parsedUrl.query else {} # 生成认证信息 auth = util.get_authorization(authRequest, \'ACS3-HMAC-SHA256\', \'\', accessKeyId, accessKeySecret) headers[\'authorization\'] = auth # 发送HTTP请求 resp = requests.post(url, body, headers=headers) # 处理响应 print(f\"状态码: {resp.status_code}\") print(f\"原始响应内容: {resp.text}\") try: response_body = resp.json() print(f\"函数调用结果: {response_body}\") except json.JSONDecodeError: response_body = resp.text print(f\"响应不是JSON格式: {response_body}\") return response_body

其实和 从 FC 调用代码是一致的,差异点就是 凭证是如何获取的

实际验证

让我们看看 HTTP 调用的实际过程:

# 设置环境变量——这个需要根据实际情况换成自己的 URLexport TARGET_FUNCTION_URL=\'https://practice-answer-tkkwptdgyr.cn-shenzhen.fcapp.run\'# 部署 httpservercaller 函数cd httpservercallers deploy# 启动本地测试s local start

使用 curl 直接访问:

curl localhost:9000/call{\"from_source\":\"answer\",\"key1\":\"http-a-val1\",\"key2\":\"http-a-val2\"}

可以看到:

  1. 通过 环境变量 成功获取了凭证
  2. 调用成功,状态码为 200
  3. 同样收到了 answer 函数的响应

常见问题和解决方案

1. 环境变量未设置

如果遇到环境变量未设置的错误:

Error Message:anonymous:1:1 >> 1| ${env.TARGET_FUNCTION_URL}RuntimeError: ${env.TARGET_FUNCTION_URL} not found

解决方案:

export TARGET_FUNCTION_URL=\'https://practice-answer-tkkwptdgyr.cn-shenzhen.fcapp.run\'
2. 权限不足

确保函数有足够的权限进行相互调用,配置正确的 RAM 角色。

总结

两种凭证获取方案的对比

  1. FC 调用context.credentials
  2. HTTP Server 调用:环境变量

调用带鉴权 HTTP 接口

# accessKeyId, accessKeySecret, securityToken 都已经获取到了# 设置请求方法和时间method = \'POST\'date = datetime.utcnow().isoformat(\'T\')[:19]+\'Z\'# 准备请求头headers = { \'x-acs-date\': date, \'x-acs-security-token\': securityToken, \'Content-Type\': \'application/json\'}# 解析URL并准备认证请求parsedUrl = urlparse(url)authRequest = TeaRequest()authRequest.method = methodauthRequest.pathname = parsedUrl.path.replace(\'$\', \'%24\')authRequest.headers = headersauthRequest.query = {k: v[0] for k, v in parse_qs(parsedUrl.query).items()} if parsedUrl.query else {}# 生成认证信息auth = util.get_authorization(authRequest, \'ACS3-HMAC-SHA256\', \'\', accessKeyId, accessKeySecret)headers[\'authorization\'] = auth# 发送HTTP请求resp = requests.post(url, body, headers=headers)

通过本文的实践,你已经掌握了在阿里云函数计算中实现有鉴权 HTTP FC 调用的完整方法。无论是使用从 FC 请求还是从 HTTP Server 请求,都能确保服务间调用的安全性和可靠性。