gdata_oauth.py

   1 #!/usr/bin/env python3
   2
   3 # https://developers.google.com/accounts/docs/OAuth2ForDevices
   4 # https://developers.google.com/drive/web/auth/web-server
   5 # https://developers.google.com/google-apps/calendar/v3/reference/calendars
   6 # https://developers.google.com/picasa-web/
   7
   8 import sys
   9 import urllib.request, urllib.parse, urllib.error
  10
  11 try:
  12     import http.client  # python2
  13 except ImportError:
  14     import http.client  # python3
  15 import os.path
  16 import json
  17 import time
  18 from typing import Dict, Optional
  19 from oauth2client.client import OAuth2Credentials  # type: ignore
  20 import gdata.calendar.service  # type: ignore
  21 import gdata.docs.service  # type: ignore
  22 import gdata.photos.service, gdata.photos  # type: ignore
  23 from googleapiclient.discovery import build  # type: ignore
  24 import httplib2  # type: ignore
  25 from googleapiclient.discovery import build
  26 import datetime
  27 import ssl
  28
  29
  30 class OAuth:
  31     def __init__(self, client_id: str, client_secret: str) -> None:
  32         print("gdata: initializing oauth token...")
  33         self.client_id = client_id
  34         self.client_secret = client_secret
  35         self.user_code: Optional[str] = None
  36         # print 'Client id: %s' % (client_id)
  37         # print 'Client secret: %s' % (client_secret)
  38         self.token: Optional[Dict] = None
  39         self.device_code = None
  40         self.verfication_url = None
  41         self.token_file = "client_secrets.json"
  42         self.scope = [
  43             #'https://www.googleapis.com/auth/calendar',
  44             #'https://www.googleapis.com/auth/drive',
  45             #'https://docs.google.com/feeds',
  46             #'https://www.googleapis.com/auth/calendar.readonly',
  47             #'https://picasaweb.google.com/data/',
  48             "https://www.googleapis.com/auth/photoslibrary.readonly",
  49             #'http://picasaweb.google.com/data/',
  50             #'https://www.google.com/calendar/feeds/',
  51         ]
  52         self.host = "accounts.google.com"
  53         self.reset_connection()
  54         self.load_token()
  55         self.last_action = 0.0
  56         self.ssl_ctx: Optional[ssl.SSLContext] = None
  57
  58     # this setup is isolated because it eventually generates a BadStatusLine
  59     # exception, after which we always get httplib.CannotSendRequest errors.
  60     # When this happens, we try re-creating the exception.
  61     def reset_connection(self) -> None:
  62         self.ssl_ctx = ssl.create_default_context(cafile="/usr/local/etc/ssl/cert.pem")
  63         http.client.HTTPConnection.debuglevel = 2
  64         self.conn = http.client.HTTPSConnection(self.host, context=self.ssl_ctx)
  65
  66     def load_token(self) -> None:
  67         token = None
  68         if os.path.isfile(self.token_file):
  69             f = open(self.token_file)
  70             json_token = f.read()
  71             self.token = json.loads(json_token)
  72             f.close()
  73
  74     def save_token(self) -> None:
  75         f = open(self.token_file, "w")
  76         f.write(json.dumps(self.token))
  77         f.close()
  78
  79     def has_token(self) -> bool:
  80         if self.token is not None:
  81             print("gdata: we have a token!")
  82         else:
  83             print("gdata: we have no token.")
  84         return self.token is not None
  85
  86     def get_user_code(self) -> Optional[str]:
  87         self.conn.request(
  88             "POST",
  89             "/o/oauth2/device/code",
  90             urllib.parse.urlencode(
  91                 {"client_id": self.client_id, "scope": " ".join(self.scope)}
  92             ),
  93             {"Content-type": "application/x-www-form-urlencoded"},
  94         )
  95         response = self.conn.getresponse()
  96         if response.status == 200:
  97             data = json.loads(response.read())
  98             self.device_code = data["device_code"]
  99             self.user_code = data["user_code"]
 100             self.verification_url = data["verification_url"]
 101             self.retry_interval = data["interval"]
 102         else:
 103             self.user_code = None
 104             print(f"gdata: {response.status}")
 105             print(response.read())
 106             sys.exit(-1)
 107         return self.user_code
 108
 109     def get_new_token(self) -> None:
 110         # call get_device_code if not already set
 111         if self.user_code is None:
 112             print("gdata: getting user code")
 113             self.get_user_code()
 114
 115         while self.token is None:
 116             self.conn.request(
 117                 "POST",
 118                 "/o/oauth2/token",
 119                 urllib.parse.urlencode(
 120                     {
 121                         "client_id": self.client_id,
 122                         "client_secret": self.client_secret,
 123                         "code": self.device_code,
 124                         "grant_type": "http://oauth.net/grant_type/device/1.0",
 125                     }
 126                 ),
 127                 {"Content-type": "application/x-www-form-urlencoded"},
 128             )
 129             response = self.conn.getresponse()
 130             if response.status == 200:
 131                 data = json.loads(response.read())
 132                 if "access_token" in data:
 133                     self.token = data
 134                     self.save_token()
 135                 else:
 136                     time.sleep(self.retry_interval + 2)
 137             else:
 138                 print("gdata: failed to get token")
 139                 print((response.status))
 140                 print((response.read()))
 141
 142     def refresh_token(self) -> bool:
 143         if self.checking_too_often():
 144             print("gdata: not refreshing yet, too soon...")
 145             return False
 146         else:
 147             print("gdata: trying to refresh oauth token...")
 148         self.reset_connection()
 149         if self.token is None:
 150             return False
 151
 152         refresh_token = self.token["refresh_token"]
 153         self.conn.request(
 154             "POST",
 155             "/o/oauth2/token",
 156             urllib.parse.urlencode(
 157                 {
 158                     "client_id": self.client_id,
 159                     "client_secret": self.client_secret,
 160                     "refresh_token": refresh_token,
 161                     "grant_type": "refresh_token",
 162                 }
 163             ),
 164             {"Content-type": "application/x-www-form-urlencoded"},
 165         )
 166
 167         response = self.conn.getresponse()
 168         self.last_action = time.time()
 169         if response.status == 200:
 170             data: Dict = json.loads(response.read())
 171             if "access_token" in data:
 172                 self.token = data
 173                 # in fact we NEVER get a new refresh token at this point
 174                 if not "refresh_token" in self.token:
 175                     self.token["refresh_token"] = refresh_token
 176                     self.save_token()
 177                 return True
 178         print(("gdata: unexpected response %d to renewal request" % response.status))
 179         print((response.read()))
 180         return False
 181
 182     def checking_too_often(self) -> bool:
 183         now = time.time()
 184         return (now - self.last_action) <= 30
 185
 186     # https://developers.google.com/picasa-web/
 187     def photos_service(self):
 188         headers = {
 189             "Authorization": "%s %s"
 190             % (self.token["token_type"], self.token["access_token"])
 191         }
 192         client = gdata.photos.service.PhotosService(additional_headers=headers)
 193         return client
 194
 195     # https://developers.google.com/drive/
 196     def docs_service(self):
 197         cred = OAuth2Credentials(
 198             self.token["access_token"],
 199             self.client_id,
 200             self.client_secret,
 201             self.token["refresh_token"],
 202             datetime.datetime.now(),
 203             "http://accounts.google.com/o/oauth2/token",
 204             "KitchenKiosk/0.9",
 205         )
 206         http = httplib2.Http(disable_ssl_certificate_validation=True)
 207         http = cred.authorize(http)
 208         service = build("drive", "v2", http)
 209         return service
 210
 211     # https://developers.google.com/google-apps/calendar/
 212     def calendar_service(self):
 213         cred = OAuth2Credentials(
 214             self.token["access_token"],
 215             self.client_id,
 216             self.client_secret,
 217             self.token["refresh_token"],
 218             datetime.datetime.now(),
 219             "http://accounts.google.com/o/oauth2/token",
 220             "KitchenKiosk/0.9",
 221         )
 222         http = httplib2.Http(disable_ssl_certificate_validation=True)
 223         http = cred.authorize(http)
 224         service = build("calendar", "v3", http)
 225         return service