Oauth coderen in een cloudfunctie met Python

Een RESTful API is een architectonisch ontwerp voor een API (Application Program Interface) die HTTP-verzoeken gebruikt om toegang te krijgen tot gegevens en deze te gebruiken. OAuth is een verificatieprotocol waarmee u een toepassing kunt goedkeuren die namens u met een andere communiceert zonder uw wachtwoord weg te geven.

OAuth is een gedelegeerd autorisatiekader waarmee apps beperkte toegang (scopes) tot de gegevens van een gebruiker kunnen verkrijgen zonder het wachtwoord van een gebruiker weg te geven. Het ontkoppelt verificatie van autorisatie en ondersteunt meerdere use cases die betrekking hebben op verschillende apparaatmogelijkheden.

De basisverificatie voor RestFul API’s lijkt ongeveer op het onderstaande codevoorbeeld, de code is gemaakt met Python en gebruikt Google Secrets Manager om de geheimen te verbergen.

Basisverificatie

    # Install Google Libary
    from google.cloud import secretmanager
    
    # Get the secret for Token
    token = _getsecrets("token")

    # Get the secret for Company ID
    companyid = _getsecrets("company_id")

    # Create the encoded authentication parameters
    auth = companyid + ':' + token
    encoded_u = base64.b64encode(auth.encode()).decode()

    # Create the Header and url
    headers = {
        "Authorization" : "Basic %s" % encoded_u
    }

Het codefragment leest de geheimen en van de geheimen maakt en codeert u de verificatie die in de koptekst wordt gebruikt. Stel de URL in en roep de functie aan _requestandstore met URL en header.

    url = 'https://api.test.com/v1/users'
    _requestandstore(url, headers)       

De functie die het verzoek, het URL-verzoek maakt, is eenvoudig door de details van de kopteksten voor het verzoek op te nemen.get

def _requestandstore(url, header):
    try:
        # Make a request
        response = requests.get(url, params={'page': 1, 'per_page': 10}, headers=header)

De volgende codefragmentfuncties zijn bedoeld om de geheimen van de Google CLoud Secrets-manager te krijgen.

def _getsecrets(secret_name):
    resource_name = f"projects/{project_id}/secrets/{secret_name}/versions/latest"
    try:
        response = client.access_secret_version(request={"name": resource_name})
        return response.payload.data.decode('UTF-8')
    except:
        return False

Oauth-verificatie

OAuth (Open Authorization) is een open standaard voor toegang tot een applicatie voor gebruikers van die applicatie om websites of applicaties toegang te verlenen tot hun informatie op andere websites, maar zonder hen de wachtwoorden te geven. Wanneer een gebruiker Oauth op een website gebruikt, verschijnt er een toestemmingsscherm om de gebruikers toestemming te geven.

Een cloudfunctie die op een serverloze server draait, heeft geen gebruikersinterface waarmee de gebruikers die toestemming kunnen geven, in het volgende deel hier zal ik uitleggen hoe we OAuth gebruiken in een cloudfunctie.

Voordat ik verder ga, vermeld ik graag dat de volgende code goed werkt in aangepaste RestFul API Oauth, als je een Google API gaat gebruiken – misschien wil je Datapipeline youtube analytics lezen naar BigQuery met Airbyte waarin ik schreef over hoe je het voor elkaar kunt krijgen met Google OAuth.

Valse Oauth Login

Om OAuth in een cloudfunctie te kunnen gebruiken, moeten we een valse OAuth-login doen, zodat we het toegangstoken kunnen krijgen, dit wordt gedaan door een ongeldig verzoek in te dienen. Als de redirect_uri ongeldig is, stopt de browser de omleiding en wordt de autorisatiecode weergegeven.

Hier is de code – eerst moeten we onze klant-ID, klantgeheim en token_url krijgen, deze informatie wordt verstrekt door de serviceprovider waarmee u verbinding probeert te maken.

De volgende stap is om de grant_type in te stellen en een RestFUL API-postverzoek in te dienen om het toegangstoken te krijgen.

    # Get the secret for Client ID
    client_id = _getsecrets("client_id")

    # Get the secret for Client Secret
    client_secret = _getsecrets("client_secret")

    # Get the secret for Token URL
    token_url = _getsecrets("token_url")

    # Fake Oauth login to get an access token by making an invalid request, If the redirect_uri is invalid,
    # the browser will stop the redirect and show the authorization code.
    data = {'grant_type': 'client_credentials'}
    access_token_response = requests.post(token_url, data=data, verify=False, allow_redirects=False, auth=(client_id, client_secret))
    tokens = json.loads(access_token_response.text)

    # Create the Header and url
    headers = {
        'Authorization': 'Bearer ' + tokens['access_token']
    }

Wanneer u het toegangstoken hebt, stelt u uw RestFUL API-URL in met eindpunt en maakt u een RestFul API-aanroep.

    url = 'https://api.test.com/api/v2/users'
    _requestandstore(filename, url, headers)

def _requestandstore(filename, url, header):
    try:
        # Make a request to API
        response = requests.get(url, params={'page': 1, 'perPage': 400}, headers=header, verify=False)
 

Dat is het, dit zou in de meeste scenario’s moeten werken – en het advies is om te praten met de serviceprovider waarmee u contact probeert te maken om de details uit te werken.

Use cases

Om te vermelden zijn er verschillende use cases waarbij u een Cloud-functie wilt gebruiken om toegang te krijgen tot een Service Providers Application. Ik zal er een paar noemen, misschien heb je meer, dus reageer hieronder.

• Service Provider extractie van gegevens naar Data Warehosue te gebruiken voor Analytics
• Social Media extractie van gegevens naar Data Warehosue te gebruiken voor Analytics
• Online Media extractie van gegevens naar Data Warehosue te gebruiken voor Analytics