Netsuite 通过 OAuth 1.0 Python 连接

Question

我正在尝试将记录发布到 Netsuite 中，它与 Postman 配合得很好，但使用 Python 时会抛出无效登录错误。

下面是代码。

import requests

url = "https://5559796-sb1.suitetalk.api.netsuite.com/services/rest/record/v1/qbc_pqr_1234"

payload="{\"aaa\":\"AN\",\"bbb\":\"OA\",\"ccc\":1,\"ddd\":false,\"eee\":114.01,\"fff\":\"OAOTWH\",\"ggg\":\"hhh\",\"hhh\":18,\"iii\":2,\"jjj\":\"2021-07-31\",\"kkk\":3257.29,\"lll\":\"Intra-Brand Stay - 3.5%\"}"
headers = {
  'Content-Type': 'application/json',
  'Authorization': 'OAuth realm="5559796_SB1",oauth_consumer_key="xxxxx",oauth_token="abcdefgh",oauth_signature_method="HMAC-SHA256",oauth_timestamp="1630905651",oauth_nonce="xxxxx",oauth_version="1.0",oauth_signature="xxxxx"'
}

response = requests.request("POST", url, headers=headers, data=payload)

print(response.text)

错误是：

{"type":"https://www.w3.org/Protocols/rfc2616/rfc2616-sec10.html#sec10.4.2","title":"Unauthorized","status":401,"o:errorDetails":[{"detail":"Invalid login attempt. For more details, see the Login Audit Trail in the NetSuite UI at Setup > Users/Roles > User Management > View Login Audit Trail.","o:errorCode":"INVALID_LOGIN"}]}

此外，如何从包含多个记录的 json 文件中读取有效负载。

Answer 1

在点击 NetSuite API 时，您需要创建一个

signature

，它负责建立您的连接，下面是对您有帮助的代码。

import hashlib
import hmac
import json
import requests
import base64
import time
import random
import urllib.parse

def _generateTimestamp():
    return str(int(time.time()))

def _generateNonce(length=11):
    """Generate pseudorandom number"""
    return ''.join([str(random.randint(0, 9)) for i in range(length)])

def _generateSignature(method, url, consumerKey, Nonce, currentTime, token, consumerSecret,
                       tokenSecret, offset):
    signature_method = 'HMAC-SHA256'
    version = '1.0'
    base_url = url
    encoded_url = urllib.parse.quote_plus(base_url)
    collected_string = None
    if type(offset) == int:
        collected_string = '&'.join(['oauth_consumer_key=' + consumerKey, 'oauth_nonce=' + Nonce,
                                     'oauth_signature_method=' + signature_method, 'oauth_timestamp=' + currentTime,
                                     'oauth_token=' + token, 'oauth_version=' + version, 'offset=' + str(offset)])
    else:
        collected_string = '&'.join(['oauth_consumer_key=' + consumerKey, 'oauth_nonce=' + Nonce,
                                     'oauth_signature_method=' + signature_method, 'oauth_timestamp=' + currentTime,
                                     'oauth_token=' + token, 'oauth_version=' + version])
    encoded_string = urllib.parse.quote_plus(collected_string)
    base = '&'.join([method, encoded_url, encoded_string])
    key = '&'.join([consumerSecret, tokenSecret])
    digest = hmac.new(key=str.encode(key), msg=str.encode(base), digestmod=hashlib.sha256).digest()
    signature = base64.b64encode(digest).decode()
    return urllib.parse.quote_plus(signature)

def import_customer():
    nsAccountID = "YOUR_ACCOUNT_ID"
    consumerKey = "YOUR_CONSUMER_KEY"
    consumerSecret = "YOUR_CONSUMER_SECRET"
    token = "YOUR_TOKEN"
    tokenSecret = "YOUR_TOKEN_SECRET"
    base_url = "https://nsAccountID.suitetalk.api.netsuite.com/services/rest/record/v1/customer"
    Nonce = self._generateNonce(length=11)
    currentTime = self._generateTimestamp()

    signature = self._generateSignature('GET', base_url, consumerKey, Nonce, currentTime, token, consumerSecret, tokenSecret, offset)

    payload = ""
    oauth = "OAuth realm=\"" + nsAccountID + "\"," \
                                                     "oauth_consumer_key=\"" + consumerKey + "\"," \
                                                                                             "oauth_token=\"" + token + "\"," \
                                                                                                                        "oauth_signature_method=\"HMAC-SHA256\"," \
                                                                                                                        "oauth_timestamp=\"" + currentTime + "\"," \
                                                                                                                                                             "oauth_nonce=\"" + Nonce + "\"," \
                                                                                                                                                                                        "oauth_version=\"1.0\"," \
                                                                                                                                                                                        "oauth_signature=\"" + signature + "\""
    headers = {
        'Content-Type': "application/json",
        'Authorization': oauth,
        'cache-control': "no-cache",
    }
    response = requests.request("GET", base_url + '?offset=' + str(offset), data=payload, headers=headers)
    return json.loads(response.text)

print(import_customer())

在这里你可以看到

import customer

的例子。在每个 API 命中中，您需要使用每个必需的信息生成签名

Answer 2

另一个简单的方法是使用 requests_oauthlib，你可以用几行让它工作并且不需要生成有点棘手的签名。

from requests_oauthlib import OAuth1
auth = OAuth1(
    client_key=CONSUMER_KEY,
    client_secret=CONSUMER_SECRET,
    resource_owner_key=TOKEN_ID,
    resource_owner_secret=TOKEN_SECRET,
    realm="YOUR_REALM",
    signature_method="HMAC-SHA256",
)
headers = {"Content-Type": "application/json"}
response = requests.request("GET", URL, auth=auth, headers=headers)