Found this content helpful? Log in or sign up to leave a like!

GraphQL authentication using browser cookies

Jump to solution
sakura_GX
Community Member

i have a python script that attempts to get assignments using the graphql API and replicating the request send when you klick the execute button (note we  neither can  nor will ever use oath2 because we do not have the resources to get verified ) 

 

#!python

import browser_cookie3
import requests
import json
cj = browser_cookie3.chromium(domain_name='iusd.instructure.com')
print(cj)

    
    

def send(cj):
    token = None
    for i in cj:
        if "<Cookie _csrf_token=" in str(i):
            token = str(i).replace("<Cookie _csrf_token=","").replace(" for iusd.instructure.com/>","")
    url = "https://iusd.instructure.com/api/graphql"


    headers = {
        "accept": "application/json+canvas-string-ids, application/json, text/plain, */*",
        "accept-language": "en-US,en;q=0.9",
        "baggage": "sentry-environment=Production,sentry-release=canvas-lms@20250409.326,sentry-public_key=355a1d96717e4038ac25aa852fa79a8f,sentry-trace_id=a62f30e1b8bc4d8bbedaa00fb431462d",
        "content-type": "application/json",
        "priority": "u=1, i",
        "sec-ch-ua": "\"Chromium\";v=\"133\", \"Not(A:Brand\";v=\"99\"",
        "sec-ch-ua-mobile": "?0",
        "sec-ch-ua-platform": "\"Linux\"",
        "sec-fetch-dest": "empty",
        "sec-fetch-mode": "cors",
        "sec-fetch-site": "same-origin",
        "sentry-trace": "",
        "x-csrf-token": token,
        "x-requested-with": "XMLHttpRequest"
    }
    # GraphQL query
    query = """
    {
        allCourses {
          name
          _id
          submissionsConnection(filter: {states: submitted}) {
            nodes {
              gradingStatus
              submissionStatus
              submittedAt
              state
              assignment {
                _id
                name
              }
              user {
                _id
                name
                sisId
                email
                enrollments {
                  state
                  course {
                    _id
                  }
                }
              }
            }
          }
          term {
            sisTermId
            name
          }
        }
    }
    """

    payload = json.dumps({
        "query": query,
        "variables": None
    })

    try:
        response = requests.post(url, headers=headers, data=payload,cookies=cj)
        print(response.request.headers)
        response.raise_for_status()  # Raises an HTTPError if the response code was unsuccessful
        print(response.json())
        return response
    except requests.exceptions.HTTPError as errh:
        print(f"HTTP Error: {errh}")
    except Exception as e:
        print(f"An error occurred: {e}")

if __name__ == "__main__":
    send(cj)
    
    
    

the current code uses  browser_cookie3 to gain all the cookies from a canvas instance (in this case a institution i am part of) and then it attempts to list assignments however  we get a 422 error.

i understand that the csrf token refreshes after each request but this was after the last request was in browser.   we need the data to create a todolist and no asking our clients to generate devkeys is not going to cut it 

HTTP Error: 422 Client Error: Unprocessable Entity for url: https://iusd.instructure.com/api/graphql

 

Labels (2)
0 Likes
1 Solution
sakura_GX
Community Member
Author

update it is working, but you need to change the headers for each browser (and set course color to None instead of attempting to fetch it from course request response since that's how it works apparently)

 

#!python

import browser_cookie3
import requests
import json
import httpx
import time
from utils import *
class api():
    def gettoken(self,cj):
        for cookiex in cj:
            if cookiex.name == '_csrf_token':
                return cookiex.value
        return None
    def hastoken(self,cj):
        return self.gettoken(cj) != None
    def send(self,url3= "/api/v1/planner/items",pg=True):
        url2 = "https://" + self.dom + url3
        #print(url2)
        token = self.t
        headers = {
            "accept": "application/json+canvas-string-ids, application/json, text/plain, */*",
            "accept-language": "en-US,en;q=0.9",
            "Content-Type": "application/json",
            "content-type": "application/json",

            "priority": "u=1, i",
            "baggage": "sentry-environment=Production,sentry-release=canvas-lms@20250409.326,sentry-public_key=355a1d96717e4038ac25aa852fa79a8f,sentry-trace_id=a62f30e1b8bc4d8bbedaa00fb431462d",

            "sec-ch-ua": "\"Chromium\";v=\"133\", \"Not(A:Brand\";v=\"99\"",
            "user-agent":"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/133.0.0.0 Safari/537.36",
            "User-Agent":"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/133.0.0.0 Safari/537.36",
            "sec-ch-ua-mobile": "?0",
            "sec-ch-ua-platform": "\"Linux\"",
            "sec-fetch-dest": "empty",
            "sec-fetch-mode": "cors",
            "sec-fetch-site": "same-origin",
            "sentry-trace": "",
            "x-csrf-token": token,
            "x-requested-with": "XMLHttpRequest"
            "authority"
        }
        # GraphQL query
        try:
            
            response = self.s.get(url2, headers=headers)
            
            #print(response.request.headers)  # Raises an HTTPError if the response code was unsuccessful
            #print(response.text)
            if response.headers != None:
                hds=dict(response.headers)
                if 'x-rate-limit-remaining' in hds:
                    if float(hds['x-rate-limit-remaining'])<400:
                        time.sleep(4)
            qp="rel=\"current\",<https://"+ self.dom
            rval=json.loads(response.content)
            if qp in str(response.headers) and "rel=\"next\"" in str(response.headers) and pg:#handle pagination
                si=str(response.headers).index(qp)
                np= str(response.headers)[si+len(qp):]
                np=np[:np.index(">;")]
                rval+=self.send(np)


                
            return rval
        except Exception as e:
            print("error"+str(e))
    def __init__(self,domain):
        self.dom = domain + ".instructure.com"
        print(self.dom)
        self.s = None
        self.cache={}
        
        d='<CookieJar[]>'
        for i in browser_cookie3.all_browsers:
            try:
                c=i(domain_name=self.dom)
                print(i)
                if (self.hastoken(c)):
                    self.t = self.gettoken(c)
                    self.s =  httpx.Client(http2=True,cookies=c)
                    
            except Exception as e:
                print(e)
                o=0
        if (self.s==None):
            raise Exception("you need to be authenticated into canvas in at least 1 browser")
        self.UUID=self.send("/api/v1/users/self")['id']

    def poll(self):
        x=self.send("/api/v1/users/self/history",pg=False)
        for i in x:
            if "assignment_" in i['asset_code']:
                if unixtime(i['visited_at']) +5 > time.time():
                    ID = i['asset_code'].replace("assignment_","")
                    if ID in self.cache and not self.cache[ID][0]:
                        sub = self.send("/api/v1/courses/"+self.cache[ID][2]+"/assignments/"+ID+"/submissions/"+self.UUID)
                        if  sub['submitted_at'] != None:
                            self.cache[ID][0] = True
                            self.traverse_assignments()
                            print("USER SUBMITTED ASSIGNMENT")
                        
            
            
        time.sleep(0.5)
    def daemon(self):
        self.traverse_assignments()
        attempts=0
        while True:
            attempts+=1
            try:
                self.poll()
                if attempts > 1600:
                    attempts=0
                    self.traverse_assignments()
            except:
                io=0
                
    def daemon2(self):
        time.sleep(90)
        while True:
            time.sleep(900)
            self.traverse_assignments()
        
        
    def traverse_assignments(self):
        c =0
        p=0
        
        for i in self.send("/api/v1/courses"):
            
            if not 'access_restricted_by_date' in i:
                print(i['id'] + " " + i['name'])
                for ix in self.send("/api/v1/courses/"+i['id']+"/assignments?bucket=upcoming") + self.send("/api/v1/courses/"+i['id']+"/assignments?bucket=ungraded"):
                    c+=1
                    color = ix['course_color']
                    if color == None:
                        color = self.send("/api/v1/users/course_"+i['id'])
                    sub = self.send("/api/v1/courses/"+i['id']+"/assignments/"+ix['id']+"/submissions/"+self.UUID)
                    if  sub['submitted_at'] == None:
                        #print("---"+ix['name'])
                        self.cache[ix['id']]=[False,ix,i['id'],color]
                        
                    else:
                        #print("P++" +ix['name'])
                        self.cache[ix['id']]=[True,ix,i['id'],color]
                        p+=1
                    
        print(100*(p/max(c,1)))
                        
                        
            
        
                
                    
                    
        
        

    








if __name__ == "__main__":
    test=api("iusd")
    test.traverse_assignments()
    while True:
        test.poll()

    
    
    
    





 

View solution in original post

0 Likes