将 Flask BasicHTTPAuth 与 Google 云函数一起使用
Using Flask BasicHTTPAuth with Google Cloud Functions
我在使用 GCP 中的 Cloud Function 时遇到困难,它只是应该 return 在使用基本 GET 请求调用时存储在 GCS Bucket 中的原始 XML。它在没有任何类型的身份验证的情况下工作正常,但是由于我将 Flask-HTTPAuth 包添加到组合中以便在公开端点之前添加一些安全措施,应用程序部署正常,但崩溃时没有任何提示为什么一旦它被调用。 SD Logging 报错如下:
severity: "DEBUG"
textPayload: "Function execution took 1847 ms, finished with status: 'crash'"
timestamp: "2020-07-15T17:22:15.158036700Z"
相关函数(匿名):
from flask import Flask, request, jsonify, make_response, abort
from flask_httpauth import HTTPBasicAuth
from google.cloud import storage, secretmanager
import google.cloud.logging
import logging
import sys
app = Flask(__name__)
auth = HTTPBasicAuth()
PROJECT_ID = 'example_project'
GCS_BUCKET = 'example_bucket'
users = ['example_user']
# Instantiate logger
client = google.cloud.logging.Client()
client.get_default_handler()
client.setup_logging()
@auth.verify_password
def verify_password(username, password):
# Instantiate the Secret Manager client.
sm_client = secretmanager.SecretManagerServiceClient()
# Load secrets
name = sm_client.secret_version_path(PROJECT_ID, 'example_secrets_ref', 1)
secrets_pass = sm_client.access_secret_version(name)
passwords = [secrets_pass]
if username in users and password in passwords:
logging.info('auth success')
return username
logging.info('auth fail')
return abort(403)
@app.route('/')
@auth.login_required
def latest_xml():
try:
request_json = request.get_json()#silent=True)
storage_client = storage.Client(project=PROJECT_ID)
bucket = storage_client.get_bucket(GCS_BUCKET)
blob = bucket.get_blob('latest_pull.xml')
latest_xml = blob.download_as_string()
logging.info('Loaded blob from GCS')
return(latest_xml)
except exception as e:
logging.error(str(e))
logging.error("Failed to load blob from GCS")
sys.exit(1)
if __name__ == '__main__':
app.run()
我试过将入口点设置为主要功能和身份验证功能都无济于事。我的问题是:是否可以在 GCP Cloud Function 中使用基本身份验证,或者我在这里找错树了?
您的函数不强制执行 standard signature for http function
def latest_xml(request):
...
这里使用的是 flask web 服务器,不需要,也没有被 Cloud Functions 使用。但是,我建议您看看 Cloud Run, and to add a simple and generic Dockerfile to deploy 。您可以在容器中部署您的“功能”as-is,并具有与 Cloud Functions 相同的行为。
编辑
当您使用 flask 时,request
object 对于每个请求都是全局的。你这样使用它:
request_json = request.get_json()#silent=True)
使用 Cloud Functions,此 object 会被 Cloud Functions 平台捕获并将参数传递给您的函数。
在 request
object 中,你有请求的 body,例如在 GET 中没用。而且,所有请求上下文:headers、用户代理、源 ip、...
我在使用 GCP 中的 Cloud Function 时遇到困难,它只是应该 return 在使用基本 GET 请求调用时存储在 GCS Bucket 中的原始 XML。它在没有任何类型的身份验证的情况下工作正常,但是由于我将 Flask-HTTPAuth 包添加到组合中以便在公开端点之前添加一些安全措施,应用程序部署正常,但崩溃时没有任何提示为什么一旦它被调用。 SD Logging 报错如下:
severity: "DEBUG"
textPayload: "Function execution took 1847 ms, finished with status: 'crash'"
timestamp: "2020-07-15T17:22:15.158036700Z"
相关函数(匿名):
from flask import Flask, request, jsonify, make_response, abort
from flask_httpauth import HTTPBasicAuth
from google.cloud import storage, secretmanager
import google.cloud.logging
import logging
import sys
app = Flask(__name__)
auth = HTTPBasicAuth()
PROJECT_ID = 'example_project'
GCS_BUCKET = 'example_bucket'
users = ['example_user']
# Instantiate logger
client = google.cloud.logging.Client()
client.get_default_handler()
client.setup_logging()
@auth.verify_password
def verify_password(username, password):
# Instantiate the Secret Manager client.
sm_client = secretmanager.SecretManagerServiceClient()
# Load secrets
name = sm_client.secret_version_path(PROJECT_ID, 'example_secrets_ref', 1)
secrets_pass = sm_client.access_secret_version(name)
passwords = [secrets_pass]
if username in users and password in passwords:
logging.info('auth success')
return username
logging.info('auth fail')
return abort(403)
@app.route('/')
@auth.login_required
def latest_xml():
try:
request_json = request.get_json()#silent=True)
storage_client = storage.Client(project=PROJECT_ID)
bucket = storage_client.get_bucket(GCS_BUCKET)
blob = bucket.get_blob('latest_pull.xml')
latest_xml = blob.download_as_string()
logging.info('Loaded blob from GCS')
return(latest_xml)
except exception as e:
logging.error(str(e))
logging.error("Failed to load blob from GCS")
sys.exit(1)
if __name__ == '__main__':
app.run()
我试过将入口点设置为主要功能和身份验证功能都无济于事。我的问题是:是否可以在 GCP Cloud Function 中使用基本身份验证,或者我在这里找错树了?
您的函数不强制执行 standard signature for http function
def latest_xml(request):
...
这里使用的是 flask web 服务器,不需要,也没有被 Cloud Functions 使用。但是,我建议您看看 Cloud Run, and to add a simple and generic Dockerfile to deploy 。您可以在容器中部署您的“功能”as-is,并具有与 Cloud Functions 相同的行为。
编辑
当您使用 flask 时,request
object 对于每个请求都是全局的。你这样使用它:
request_json = request.get_json()#silent=True)
使用 Cloud Functions,此 object 会被 Cloud Functions 平台捕获并将参数传递给您的函数。
在 request
object 中,你有请求的 body,例如在 GET 中没用。而且,所有请求上下文:headers、用户代理、源 ip、...