0

我正在尝试从用 python 编写的谷歌云函数中使用谷歌 cloudiot api。当我向 Cloud IoT api 发出 http 请求时，我收到 403“禁止”错误。我已使用具有完全权限的服务帐户进行了身份验证。

从烧瓶导入 os 导入 json 导入 base64

从 firebase_admin import db 导入 firebase_admin

从 googleapiclient 导入 googleapiclient 从 google.oauth2 导入发现导入 service_account

default_app = firebase_admin.initialize_app(options={'databaseURL': ' https://lucid-v2-1.firebaseio.com/ '})

def get_client(service_account_json): """通过发现 IoT API 并使用服务帐户凭据 JSON 创建服务对象来返回授权的 API 客户端。""" api_scopes = [' https://www.googleapis.com/auth/云平台', ' https://www.googleapis.com/auth/cloudiot '] api_version = 'v1' discovery_api = ' https://cloudiot.googleapis.com/ $discovery/rest' service_name = 'cloudiotcore'

#credentials = service_account.Credentials.from_service_account_file(service_account_json)
credentials = service_account.Credentials.from_service_account_file(service_account_json)
scoped_credentials = credentials.with_scopes(api_scopes)

discovery_url = '{}?version={}'.format(
        discovery_api, api_version)

print(credentials.service_account_email)

return discovery.build(
        service_name,
        api_version,
        discoveryServiceUrl=discovery_url,
        credentials=scoped_credentials,
        cache_discovery=False)

def request_posted（事件，上下文）：

print(os.environ)
#
client = get_client(os.environ.get('GOOGLE_APPLICATION_CREDENTIALS'))
"""Triggered by a change to a Firebase RTDB reference.
Args:
     event (dict): Event payload.
     context (google.cloud.functions.Context): Metadata for the event.

"""

print(client)

resources = context.resource.split('/')
request_id = resources[-1]
request_values = event['delta']

response_ref = db.reference("experience_channel/responses/" + str(request_id))
response_ref.update({"ack":True})

db.reference("experience_channel/experience_data")

device_id = "2593786060198708"
project_id = "lucid-iOS-v2-1"
cloud_region = "us-central1"
registry_id = "Lucid_IoT_Registry"
device_path = 'projects/{}/locations/{}/registries/{}/devices/{}'.format(project_id, cloud_region, registry_id, device_id)

print(device_path)
epoch_length = 20

message_contents = {
"experience_id":request_id,
"epoch_length":epoch_length,
"experience_length":request_values['length']
}

message_json = json.dumps(message_contents)

test_body = "test_body"

command_body = {
    'binaryData': base64.urlsafe_b64encode(
        message_json.encode('utf-8')).decode('ascii'),
    'subfolder' : 'commands'
}

#client.projects().locations().registries().testIamPermissions(resource = )


print(client.projects(
    ).locations().registries(
    ).devices().list(parent = "projects/lucid-iOS-v2-1/locations/us-central1/registries/Lucid_IoT_Registry").execute())

resp = client.projects(
    ).locations().registries(
    ).devices().sendCommandToDevice(
    name = device_path, body = command_body).execute()

print(resp)

return resp
4

0 回答 0