Skip to content

Instantly share code, notes, and snippets.

@TheOnlyWayUp
Last active June 25, 2022 07:47
Show Gist options
  • Star 1 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save TheOnlyWayUp/6106dec0e5974026aa7144ab6761b0c6 to your computer and use it in GitHub Desktop.
Save TheOnlyWayUp/6106dec0e5974026aa7144ab6761b0c6 to your computer and use it in GitHub Desktop.
I was having some trouble with Zoho's OAuth, after lots of research, this is what I came up with. The goal of using Zoho's OAuth is mostly to verify that the person logging in is from a specific domain..

Zoho OAuth Identification Example

The goal of this snippet is to help you figure out the email/identity of the person OAUthing to your application. Made this to help out anyone else that's trying to do the same thing.

main.py - Just a short aiohttp snippet regarding what requests you need to make to get the access token, and the user's information.

example_usage.py - A FastAPI implementation of this.

NOTE = You need to select the server based applications option when creating your Application at Zoho's Console. Then you need to run a webserver, add that server's domain (you need to have a domain iirc) as an authorized redirect url. Think your_domain.com/callback.

How to use

  • Make an application at Zoho's console (url above)
  • select the server based application option
  • set the name and such, copy the client ID and client secret to the config
  • paste your domain name in the place of your_domain.com (complete url in the config should be your_domain.com/callback) in the config
  • Go to zoho's console and add an Authorized Redirect URL, as https://your_domain.com/callback
  • Run your webserver (python3.10 example_usage.py)
  • Visit your_domain.com, you'll be redirected to Zoho's oauth, once you allow it, it'll send a callback request to your redirect URL, which processes everything, sends two more requests and finally serves you your email address.

Ideally, instead of just returning the email address, you should verify it, add it to a database, hash it and set that as a cookie, and use that to authenticate on every request. Either using a FastAPI Middleware or Dependencies. Then, once a user reaches the callback URL, after all this processing and setting cookies, return a redirect response back to your home page (which should check for the presence of cookies ideally and serve/redirect appropriately).

Extra

This isn't really important to actually using zoho oauth, but for general login type beat, this could be useful -

Make sure to check out the callback route, it sets the cookie, adds it to the database, etc. Also, in my setup, the / route checks for appropriate cookies, verifies them and redirects to the correct page based on that. (Not logged in/invalid cookie: redirect to login page, logged in: redirect to status page), that's why everything redirects to /.

from fastapi import FastAPI, Cookie, Depends, Response, Request
from typing import Tuple, Optional
import aioredis

config = {
        "redirect_url": "your_domain.com/callback",
        "client_id": "your_client_id_here",
        "client_secret": "client_secret",
        "scope": "ZohoMail.accounts.READ"
    }

redis = aioredis.from_url("redis://localhost") # Make sure to install `redis-server` and nohup it (`nohup redis-server &`) before trying to connect to the database

async def validate_session(authorization: str = Cookie(default=None)) -> Tuple[bool, Optional[dict]]:
    if authorization is None:
        return False, None

    data = await redis.hmget(authorization, ["email_address", "access_token"])
    
    if data is None:
        return False, None
    
    to_return = {
        "email_address": data[0],
        "access_token": data[1]
    }
    return True, to_return

@app.get("/send")
def send_frontend(request: Request, auth_status = Depends(validate_session)):
    if auth_status[0] is False:
        return RedirectResponse("/")
    return Response(content=auth_status[1], status_code=200)

@app.get("/callback")
async def oauth_redir(request: Request, code: str, location: str, authorization: str = Cookie(default=None)):
    
    if authorization is not None:
        token = await redis.hmget(authorization, ["email_address", "access_token"])
        if token is not None:
            return RedirectResponse("/", headers={"reason": "Already logged in"})


    async with aiohttp.ClientSession() as session:
        async with session.post(f"https://accounts.zoho.com/oauth/v2/token?code={code}&grant_type=authorization_code&client_id={config['client_id']}&client_secret={config['client_secret']}&redirect_uri={config['redirect_url']}&scope={config['scope']}") as response:
            data = await response.json()
            access_token = data['access_token']
            ttl = data['expires_in']
        
        async with session.get("https://mail.zoho.com/api/accounts", headers={"Authorization": f"Zoho-oauthtoken {access_token}"}) as response:
            data = await response.json()
    data = data['data'][0]

    email_address = data['mailboxAddress']
    
    cookie_key = str(uuid4())
    
    await redis.hmset(cookie_key, {"access_token": access_token, "email_address": email_address})
    await redis.expire(cookie_key, int(ttl))
    
    r_response = RedirectResponse("/")
    r_response.set_cookie("authorization", cookie_key, expires = int(ttl))
    return r_response
import aiohttp
from fastapi import FastAPI, Request
from fastapi.responses import RedirectResponse
config = {
"redirect_url": "your_domain.com/callback",
"client_id": "your_client_id_here",
"client_secret": "client_secret",
"scope": "ZohoMail.accounts.READ"
}
app = FastAPI()
@app.get("/")
def main():
return RedirectResponse(f"https://accounts.zoho.com/oauth/v2/auth?scope={config['scope']}&client_id={config['client_id']}&response_type=code&access_type=offline&redirect_uri={config['redirect_url']}")
@app.get("/callback")
async def oauth_redir(request: Request, code: str, location: str):
async with aiohttp.ClientSession() as session:
async with session.post(f"https://accounts.zoho.com/oauth/v2/token?code={code}&grant_type=authorization_code&client_id={config['client_id']}&client_secret={config['client_secret']}&redirect_uri={config['redirect_url']}&scope={config['scope']}") as response:
data = await response.json()
access_token = data['access_token']
ttl = data['expires_in']
async with session.get("https://mail.zoho.com/api/accounts", headers={"Authorization": f"Zoho-oauthtoken {access_token}"}) as response:
data = await response.json()
data = data['data'][0]
email_address = data['mailboxAddress']
return email_address
import aiohttp
config = {
"redirect_url": "your_domain.com/callback",
"client_id": "your_client_id_here",
"client_secret": "client_secret",
"scope": "ZohoMail.accounts.READ"
}
async def main(code: str):
async with aiohttp.ClientSession() as session:
async with session.post(f"https://accounts.zoho.com/oauth/v2/token?code={code}&grant_type=authorization_code&client_id={config['client_id']}&client_secret={config['client_secret']}&redirect_uri={config['redirect_url']}&scope={config['scope']}") as response:
data = await response.json()
access_token = data['access_token']
ttl = data['expires_in']
async with session.get("https://mail.zoho.com/api/accounts", headers={"Authorization": f"Zoho-oauthtoken {access_token}"}) as response:
data = await response.json()
return data
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment