How to Capture The OAuth Redirect to POST A Response

Question

So my colleague and I have an application whereby we need to capture the OAuth Redirect from Google's OAuth Server Response, the reason being is we need to send a payload to capture to renew our pickle token, and we need to do it without human intervention. The code looks like this:

#!/usr/bin/env python3
import pickle
import os.path
import pandas as pd
import requests
from googleapiclient.discovery import build
from google_auth_oauthlib.flow import InstalledAppFlow
from google.auth.transport.requests import Request
import base64
from datetime import datetime, timedelta
from urllib.parse import unquote
from bs4 import BeautifulSoup


# If modifying these scopes, delete the file token.pickle.
SCOPES = ['https://www.googleapis.com/auth/gmail.readonly']


def search_message(service, user_id, search_string):
    """
        Search the inbox for emails using standard gmail search parameters
        and return a list of email IDs for each result
        PARAMS:
            service: the google api service object already instantiated
            user_id: user id for google api service ('me' works here if
            already authenticated)
            search_string: search operators you can use with Gmail
            (see https://support.google.com/mail/answer/7190?hl=en for a list)
        RETURNS:
            List containing email IDs of search query
        """


    try:
        # initiate the list for returning
        list_ids = []

        # get the id of all messages that are in the search string
        search_ids = service.users().messages().list(userId=user_id, q=search_string).execute()

        # if there were no results, print warning and return empty string
        try:
            ids = search_ids['messages']

        except KeyError:
            print("WARNING: the search queried returned 0 results")
            print("returning an empty string")
            return ""

        if len(ids) > 1:
            for msg_id in ids:
                list_ids.append(msg_id['id'])
            return (list_ids)

        else:
            list_ids.append(ids['id'])
        return list_ids

    except:
        print("An error occured: %s")


def get_message(service, user_id, msg_id):
    """
        Search the inbox for specific message by ID and return it back as a
        clean string. String may contain Python escape characters for newline
        and return line.

        PARAMS
            service: the google api service object already instantiated
            user_id: user id for google api service ('me' works here if
            already authenticated)
            msg_id: the unique id of the email you need
        RETURNS
            A string of encoded text containing the message body
        """


    try:
        final_list = []

        message = service.users().messages().get(userId=user_id, id=msg_id).execute()  # fetch the message using API
        payld = message['payload']  # get payload of the message


        report_link = ""
        mssg_parts = payld['parts']  # fetching the message parts
        part_one = mssg_parts[1]  # fetching first element of the part
        #part_onee = part_one['parts'][1]
        #print(part_one)
        part_body = part_one['body']  # fetching body of the message
        part_data = part_body['data']  # fetching data from the body
        clean_one = part_data.replace("-", "+")  # decoding from Base64 to UTF-8
        clean_one = clean_one.replace("_", "/")  # decoding from Base64 to UTF-8
        clean_one = clean_one.replace("amp;", "")  # cleaned amp; in links
        clean_two = base64.b64decode(clean_one)  # decoding from Base64 to UTF-8
        soup = BeautifulSoup(clean_two, features="html.parser")
        for link in soup.findAll('a'):
            if "talentReportRedirect?export" in link.get('href'):
                report_link = link.get('href')
                break
        final_list.append(report_link)  # This will create a dictonary item in the final list
    except Exception:
        print("An error occured: %s")
    return final_list


def get_service():
    """
        Authenticate the google api client and return the service object
        to make further calls
        PARAMS
            None
        RETURNS
            service api object from gmail for making calls
        """


    creds = None

    # The file token.pickle stores the user's access and refresh tokens, and is
    # created automatically when the authorization flow completes for the first
    # time.
    if os.path.exists('token.pickle'):
        with open('token.pickle', 'rb') as token:
            creds = pickle.load(token)

    # If there are no (valid) credentials available, let the user log in.
    if not creds or not creds.valid:
        if creds and creds.expired and creds.refresh_token:
            creds.refresh(Request())
        else:
            flow = InstalledAppFlow.from_client_secrets_file('credentials.json', SCOPES)
            creds = flow.run_local_server(port=0)

    # Save the credentials for the next run
    with open('token.pickle', 'wb') as token:
        pickle.dump(creds, token)

        auth_link = build('gmail', 'v1', credentials=creds)
        parsed_url = unquote(auth_link).split('redirect')[-1]
        return parsed_url


def get_report(link_array):
    for link in link_array:
       df = requests.get(link[0], allow_redirects=True)
       # df.encoding
       # dt = pd.DataFrame(data=df)
       print(link)
       # upload_to_database(df)  -- Richard Barret please update this function
       print(df.text)

      ## dt.to_csv(r'C:\Users\user\Desktop\api_gmail.csv', sep='\t',header=True)

if __name__ == "__main__":
    link_list = []
    monday = datetime(2022,12,5)#datetime.now() - timedelta(days=datetime.now().weekday())
    thursday = datetime(2022,12,8)#datetime.now() - timedelta(days=datetime.now().weekday() - 3)
    query = 'from:[email protected] ' + 'after:' + monday.strftime('%Y/%m/%d') + ' before:' + thursday.strftime('%Y/%m/%d')
    service = get_service()
    mssg_list = search_message(service, user_id='me', search_string=query)
    for msg in mssg_list:
        link_list.append(get_message(service, user_id='me', msg_id=msg))
    get_report(link_list)

It is assumed that you have a directory structure like this:

├── credentials.json
├── gmail_api_linkedin.py
└── requirements.txt

Obviously, you won't have the credentials.json file, but in essence, the code works and redirects us to a login page to retrieve the new pickle:

The main thing is we can't interact with that in an autonomous fashion. As such, how can we capture the URL from the server that prints out the following information the is differenter every single time.

Please visit this URL to authorize this application: https://accounts.google.com/o/oauth2/auth?response_type=code&client_id=212663976989-96o952s9ujadjgfdp6fm0p462p37opml.apps.googleusercontent.com&redirect_uri=http%3A%2F%2Flocalhost%3A58605%2F&scope=https%3A%2F%2Fwww.googleapis.com%2Fauth%2Fgmail.readonly&state=ztJir0haFQlvTP79BRthhmEHlSsqIj&access_type=offline

More succinctly, how can we capture the URL in a pythonic manner to send POST and PUT requests to that redirect?

Answer 1

renew our pickle token

I still do not understand why you feel the need to renew your token pickle.

how it all works.

The following example will spawn the consent screen directly on the machine its running on. It then stores the token within the token.json file

token.json

This file contains all the information needed by the script to run. It can automatically request a new access token when ever it needs.

{
  "token": "[REDACTED]",
  "refresh_token": "[REDACTED]",
  "token_uri": "https://oauth2.googleapis.com/token",
  "client_id": "[REDACTED]",
  "client_secret": "[REDACTED],
  "scopes": [
    "https://mail.google.com/"
  ],
  "expiry": "2023-01-03T19:06:13.959468Z"
}

gmail quickstart.

#   To install the Google client library for Python, run the following command:
#   pip install --upgrade google-api-python-client google-auth-httplib2 google-auth-oauthlib


from __future__ import print_function

import os.path

import google.auth.exceptions
from google.auth.transport.requests import Request
from google.oauth2.credentials import Credentials
from google_auth_oauthlib.flow import InstalledAppFlow
from googleapiclient.discovery import build
from googleapiclient.errors import HttpError

# If modifying these scopes, delete the file token.json.
SCOPES = ['https://mail.google.com/']

def main():

    """Shows basic usage of the Gmail v1 API.
    Prints a list of user messages.
    """
    creds = None
    # The file token.json stores the user's access and refresh tokens, and is
    # created automatically when the authorization flow completes for the first
    # time.
    if os.path.exists('token.json'):
        try:
            creds = Credentials.from_authorized_user_file('token.json', SCOPES)
            creds.refresh(Request())
        except google.auth.exceptions.RefreshError as error:
            # if refresh token fails, reset creds to none.
            creds = None
            print(f'An error occurred: {error}')
    # If there are no (valid) credentials available, let the user log in.
    if not creds or not creds.valid:
        if creds and creds.expired and creds.refresh_token:
            creds.refresh(Request())
        else:
            flow = InstalledAppFlow.from_client_secrets_file(
                'C:\YouTube\dev\credentials.json', SCOPES)
            creds = flow.run_local_server(port=0)
        # Save the credentials for the next run
        with open('token.json', 'w') as token:
            token.write(creds.to_json())

    try:
        service = build('gmail', 'v1', credentials=creds)

        # Call the Gmail v1 API
        results = service.users().messages().list(
            userId='me').execute()
        messages = results.get('messages', [])

        if not messages:
            print('No messages found.')
            return
        print('Messages:')
        for message in messages:
            print(u'{0} ({1})'.format(message['id'], message['threadId']))
    except HttpError as error:
        # TODO(developer) - Handle errors from gmail API.
        print(f'An error occurred: {error}')


if __name__ == '__main__':
    main()

expired refresh token.

If your issue is in fact that your refresh tokens are expiring this is because your app is currently in the testing phase. If you set your app to production then your refresh tokens will stope expiring.