chris/dotfiles
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

adding oauth script

Browse source
This commit is contained in:
Chris Cochrun 2024-01-03 10:34:22 -06:00
parent 6534c2b145
commit 2d11ded544
2 changed files with 717 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

296
.config/tridactyl/tridactylrc-copied Normal file
View file

@ -0,0 +1,296 @@
alias tabsort jsb browser.tabs.query({}).then(tabs => tabs.sort((t1, t2) => t1.url.localeCompare(t2.url)).forEach((tab, index) => browser.tabs.move(tab.id, {index})))
alias tabuniq jsb browser.tabs.query({}).then(tabs => browser.tabs.remove(tabs.filter((tab, index) => tabs.slice(index + 1).find(t => t.url == tab.url)).map(tab => tab.id)))
alias logall js let l=prompt() ; Object.keys(tri.config.get("logging")).forEach(k => tri.config.set("logging", k, l))
alias playAllVideos js tri.native.run("mpv --really-quiet --ontop --keepaspect-window --profile=protocol.http " + Array.from(document.querySelectorAll("a, iframe, video")).reduce((s, e) => {let r=(/^https?:\/\/((www.)?youtu((\.be\/)|(be\.com\/((embed\/)|(watch\?v=))))[^ ]+)|(.+\.webm)$/);let l="";if(e.tagName=="IFRAME")l=e.src.match(r);else if(e.tagName=="A")l=e.href.match(r)||e.innerText.match(r);else if(e.tagName=="VIDEO")l=[e.currentSrc?e.currentSrc:e.src];console.log(l);return s+(l && l.length > 0 && s.indexOf(l[0])<0?"'"+l[0]+"' ":"")},""))
alias gitclone jsb -p tri.native.run("git clone --depth=1 '" + JS_ARG + "' /home/me/prog/" + JS_ARG.split("/").slice(-1))
alias rsssave jsb -p tri.native.run('cat >> ~/.config/newsboat/urls', JS_ARG + "\n")
alias openGithubNotifications composite js Array.from(document.querySelectorAll("li.list-group-item > span:nth-child(1) > a:nth-child(2)")).map(e => e.href) | jsb -p JS_ARG.forEach(url => tri.excmds.tabopen(url))
alias tn jsb let a = str.join("").split("/" + "/").slice(-1)[0]; tri.excmds.tabopen(a ? `https://intranet.adacore.com/crm/#/tn/${a}` : `https://intranet.adacore.com/crm/#/Tickets/myTNs`) //
alias mktn js tri.native.run(`mkdir -v "$HOME/prog/${document.location.href.split("/").slice(-1)[0]}"`).then(r=>Array.from(document.querySelectorAll("#Files a")).forEach(a=>fetch(a.href).then(r=>r.text()).then(t=>tri.native.write(`${r.content.match(/'.*'\n/)[0].slice(1,-2)}/${a.href.split("/").slice(-1)[0]}`,t))))
autocmd DocLoad twitter.com urlmodify -t twitter.com nitter.net
bind O composite url2args | fillcmdline open
bind gh followpage prev
bind gl followpage next
bind gc composite js window.location.href | gitclone
bind ;gc hint -qW gitclone
bind H tabprev
bind L tabnext
bind K forward
bind J back
bind y clipboard yankshort
bind Y hint -p
bind v composite hint -pipe a href | js -p tri.excmds.shellescape(JS_ARG) | exclaim_quiet mpv --ontop --keepaspect-window --profile=protocol.http
bind V js tri.excmds.shellescape(document.location.href).then(url => tri.native.run(`mpv --ontop --keepaspect-window --profile=protocol.http '${url}'`))
bind ;v composite hint -qpipe a href | js -p JS_ARG.map(h => `'${h}'`).join(" ") | ! mpv
bind e hint -W js -p tri.native.run(`$HOME/bin/add-magnet '${JS_ARG}'`)
bind u undo tab
bind U undo window
bind s fillcmdline saveas
bind S saveas
bind ;s hint -a
bind ;S hint -s
bind <A-v> playAllVideos
" Bind <A-&..à> to `buffer 1..$` and <A-1..0> to `tabmove 1..$`
jsb ["&", "é", '"', "'", "(", "-", "è", "_", "ç", "à"].forEach((l, i) => { i = (i == 9 ? 0 : (i + 1)); ["--mode=insert", "--mode=input", "--mode=normal"].forEach(mode => { tri.excmds.bind(mode, `<A-${l}>`, `buffer ${i}`) ; tri.excmds.bind(mode, `<A-${i}>`, `tabmove ${i}`) })})
jsb ["--mode=insert", "--mode=input", "--mode=normal"].forEach(mode => { tri.excmds.bind(mode, `<A-H>`, `tabmove -1`) ; tri.excmds.bind(mode, `<A-L>`, `tabmove +1`) ; })
bind --mode=normal <C-P> winopen -private
bind --mode=ex <C-a> text.beginning_of_line
bind --mode=insert <C-a> text.beginning_of_line
bind --mode=input <C-a> text.beginning_of_line
bind --mode=ex <C-e> text.end_of_line
bind --mode=insert <C-e> text.end_of_line
bind --mode=input <C-e> text.end_of_line
bind --mode=ex <C-f> text.forward_word
bind --mode=insert <C-f> text.forward_word
bind --mode=input <C-f> text.forward_word
bind --mode=ex <C-k> text.kill_line
bind --mode=insert <C-k> text.kill_line
bind --mode=input <C-k> text.kill_line
bind --mode=ex <C-u> text.backward_kill_line
bind --mode=insert <C-u> text.backward_kill_line
bind --mode=input <C-u> text.backward_kill_line
bind --mode=ex <C-V> composite getclip selection | text.insert_text
bind --mode=insert <C-V> composite getclip selection | text.insert_text
bind --mode=input <C-V> composite getclip selection | text.insert_text
bind --mode=ex <C-w> text.backward_kill_word
bind --mode=insert <C-w> text.backward_kill_word
bind --mode=input <C-w> text.backward_kill_word
" Requires custom firefox build: https://github.com/glacambre/firefox-patches
bind --mode=ex <C-n> ex.next_completion
bind --mode=ex <C-p> ex.prev_completion
bind --mode=ex <C-y> ex.insert_completion
bind --mode=ex <C-e> ex.deselect_completion
bind --mode=ex <Tab> ex.complete
bind --mode=ex <C-g> composite text.beginning_of_line ; text.forward_word ; text.kill_word
" Disable <C-q>
bind --mode=insert <C-q> js alert("<C-q> is for quitters.")
bind --mode=input <C-q> js alert("<C-q> is for quitters.")
bind --mode=normal <C-q> js alert("<C-q> is for quitters.")
bind --mode=ex <C-q> js alert("<C-q> is for quitters.")
bind --mode=hint <C-q> js alert("<C-q> is for quitters.")
unbind yy
unbind ys
unbind yc
unbind ym
unbind yt
unbind --mode=ex <C-c>
unbind --mode=ex <Space>
unbind <SA-ArrowUp><SA-ArrowUp><SA-ArrowDown><SA-ArrowDown><SA-ArrowLeft><SA-ArrowRight><SA-ArrowLeft><SA-ArrowRight>ba
bindurl reddit.com <Space><Space> urlmodify -t www old
bindurl https://github.com/.*/.*/blob f hint -c .blob-num,a
bindurl https://github.com/notifications <Space><Space> openGithubNotifications
bindurl youtu((\.be)|(be\.com)) f hint -J
bindurl google(\.[a-zA-Z0-9]+){1,2}/search f hint -Jc #top_nav a, #search a, .card-section a, a.fl, #pnnext, #pnprev
bindurl google(\.[a-zA-Z0-9]+){1,2}/search F hint -Jbc #top_nav a, #search a, .card-section a, a.fl, #pnnext, #pnprev
bindurl google(\.[a-zA-Z0-9]+){1,2}/search gF hint -Jqbc #top_nav a, #search a, .card-section a, a.fl, #pnnext, #pnprev
bindurl lkml.org/lkml gl js let lis = Array.from(document.querySelectorAll(".threadlist:nth-of-type(1)")[0].querySelectorAll("li")); document.location.href = lis[lis.findIndex(li => li.className.match("origin")) + 1].querySelector("a").href
bindurl lkml.org/lkml gh js let lis = Array.from(document.querySelectorAll(".threadlist:nth-of-type(1)")[0].querySelectorAll("li")); document.location.href = lis[lis.findIndex(li => li.className.match("origin")) - 1].querySelector("a").href
set allowautofocus false
set hintchars fdsqjklmrezauiopwxcvghtybn
set searchengine g
set tabopencontaineraware true
set rsscmd rsssave %u
set visualenterauto false
set wordpattern [^\s\/]+
" Disable all searchurls
jsb Object.keys(tri.config.get("searchurls")).reduce((prev, u) => prev.catch(()=>{}).then(_ => tri.excmds.setnull("searchurls." + u)), Promise.resolve())
" Add our own
set searchurls.amazon https://www.amazon.fr/s/ref=nb_sb_noss?field-keywords=%s
set searchurls.bandcamp https://bandcamp.com/search?q=%s
set searchurls.cnrtl http://www.cnrtl.fr/lexicographie/%s
set searchurls.conj http://www.les-verbes.com/conjuguer.php?verbe=%s
set searchurls.crates https://crates.io/search?q=%s
set searchurls.ddg https://duckduckgo.com/html?q=%s
set searchurls.deb https://packages.debian.org/search?keywords=%s&searchon=names&suite=all&section=all
set searchurls.fdroid https://search.f-droid.org/?q=%s
set searchurls.g https://www.google.com/search?q=%s
set searchurls.gh https://github.com/search?utf8=%E2%9C%93&q=%s&ref=simplesearch
set searchurls.gi https://www.google.com/search?q=%s&tbm=isch
set searchurls.gmaps https://www.google.com/maps/search/%s
set searchurls.gw https://wiki.gentoo.org/index.php?title=Special%3ASearch&profile=default&search=%s&fulltext=Search
set searchurls.imdb https://www.imdb.com/find?q=%s
set searchurls.lqwant https://lite.qwant.com/?q=%s
set searchurls.mdn https://developer.mozilla.org/en-US/search?q=%s&topic=api&topic=js
set searchurls.monova https://monova.to/search?term=%s
set searchurls.npm https://www.npmjs.com/search?q=%s
set searchurls.osm https://www.openstreetmap.org/search?query=%s
set searchurls.pydoc https://docs.python.org/3/search.html?q=%s
set searchurls.qwant https://www.qwant.com/?q=%s
set searchurls.ratp https://www.ratp.fr/itineraires?start=%s1&end=%s2&lieu_depart=&lieu_arrivee=&modes[rail]=rail&modes[metro]=metro&modes[bus]=bus&modes[tram]=tram&itinerary_profile=fastest&op=C%27est+parti
set searchurls.r https://old.reddit.com/r/%s
set searchurls.rustdoc https://doc.rust-lang.org/std/index.html?search=%s
set searchurls.searxme https://searx.me/?q=%s&categories=general&language=en-US
set searchurls.skyt https://www.skytorrents.to/?search=%s
set searchurls.steam https://store.steampowered.com/search/?term=%s
set searchurls.torrentz https://torrentz2.eu/search?f=%s
set searchurls.tpb https://thepiratebay.org/s/?q=%s&=on&page=0&orderby=99
set searchurls.tre http://www.wordreference.com/redirect/translation.aspx?w=%s&dict=enfr
set searchurls.trf http://www.wordreference.com/redirect/translation.aspx?w=%s&dict=fren
set searchurls.w https://en.wikipedia.org/w/index.php?search=%s&title=Special%3ASearch
set searchurls.wfr https://fr.wikipedia.org/w/index.php?search=%s&title=Sp%E9cial%3ARecherche
set searchurls.y https://www.youtube.com/results?search_query=%s
set customthemes.custom html, body { -moz-font-feature-settings: "dlig" 0 !important; } span.TridactylHint { font-family: monospace !important; background: transparent !important; color: black !important; text-shadow: cyan -1px -1px 0px, cyan -1px 0px 0px, cyan -1px 1px 0px, cyan 1px -1px 0px, cyan 1px 0px 0px, cyan 1px 1px 0px, cyan 0px 1px 0px, cyan 0px -1px 0px !important; }
set theme custom
seturl jsfiddle.net allowautofocus true
" Native messenger stuff
guiset_quiet hoverlink right
guiset_quiet tabs count
setpref accessibility.typeaheadfind.autostart false
setpref accessibility.typeaheadfind.flashBar 0
setpref app.normandy.api_url ""
setpref app.normandy.enabled false
setpref app.normandy.first_run false
setpref app.shield.optoutstudies.enabled false
setpref app.update.enabled false
setpref beacon.enabled false
setpref beacon.enabled false
setpref browser.autofocus false
setpref browser.aboutHomeSnippets.updateUrl "data:,"
setpref browser.display.use_document_fonts 0
setpref browser.download.dir "/home/me/downloads"
setpref browser.download.folderList 2
setpref browser.download.manager.addToRecentDocs false
setpref browser.download.useDownloadDir false
setpref browser.eme.ui.enabled false
setpref browser.feeds.handler.default "client"
setpref browser.feeds.handlers.application "/home/me/bin/add_rss_feed"
setpref browser.formfill.enable false
setpref browser.helperApps.deleteTempFileOnExit true
setpref browser.library.activity-stream.enabled false
setpref browser.messaging-system.whatsNewPanel.enabled false
setpref browser.newtab.preload false
setpref browser.newtab.url "about:blank"
setpref browser.newtabpage.enabled false
setpref browser.newtabtabpage.enabled false
setpref browser.newtabtabpage.enhanced false
setpref browser.onboarding.enabled false
setpref browser.pagethumbnails.capturing_disabled true
setpref browser.ping-centre.telemetry false
setpref browser.pocket.api ""
setpref browser.pocket.oAuthConsumerKey ""
setpref browser.pocket.site ""
setpref browser.safebrowsing.appRepURL ""
setpref browser.safebrowsing.blockedURIs.enabled false
setpref browser.safebrowsing.downloads.remote.enabled false
setpref browser.safebrowsing.downloads.remote.url ""
setpref browser.safebrowsing.gethashURL ""
setpref browser.safebrowsing.malware.enabled false
setpref browser.safebrowsing.malware.reportURL ""
setpref browser.safebrowsing.phishing.enabled false
setpref browser.safebrowsing.provider.google.gethashURL ""
setpref browser.safebrowsing.provider.google.lists ""
setpref browser.safebrowsing.provider.google.reportMalwareMistakeURL ""
setpref browser.safebrowsing.provider.google.reportPhishMistakeURL ""
setpref browser.safebrowsing.provider.google.reportURL ""
setpref browser.safebrowsing.provider.google.updateURL ""
setpref browser.safebrowsing.provider.google4.dataSharing.enabled false
setpref browser.safebrowsing.provider.google4.dataSharingURL ""
setpref browser.safebrowsing.provider.google4.reportMalwareMistakeURL ""
setpref browser.safebrowsing.provider.google4.reportPhishMistakeURL ""
setpref browser.safebrowsing.provider.google4.reportURL ""
setpref browser.safebrowsing.provider.mozilla.gethashURL ""
setpref browser.safebrowsing.provider.mozilla.updateURL ""
setpref browser.safebrowsing.reportPhishURL ""
setpref browser.safebrowsing.reportURL ""
setpref browser.safebrowsing.updateURL ""
setpref browser.search.region "US"
setpref browser.search.suggest.enabled false
setpref browser.send_pings false
setpref browser.send_pings.require_same_host true
setpref browser.sessionstore.restore_on_demand false
setpref browser.shell.checkDefaultBrowser false
setpref browser.startup.homepage "about:blank"
setpref browser.startup.homepage_override.mstone "ignore"
setpref browser.startup.page 3
setpref browser.tabs.closeWindowWithLastTab false
setpref browser.tabs.remote.autostart.2 true
setpref browser.uidensity 1
setpref browser.urlbar.placeholderName ""
setpref browser.urlbar.trimURLs false
setpref datareporting.healthreport.uploadEnabled false
setpref datareporting.policy.dataSubmissionEnabled false
setpref devtools.gcli.hideIntro true
setpref devtools.scratchpad.enabled true
setpref devtools.scratchpad.wrapText true
setpref devtools.webide.autoinstallADBHelper false
setpref devtools.webide.enabled false
setpref extensions.formautofill.addresses.enabled false
setpref extensions.formautofill.available "off"
setpref extensions.formautofill.creditCards.enabled false
setpref extensions.formautofill.heuristics.enabled false
setpref extensions.pocket.enabled false
setpref extensions.screenshots.disabled true
setpref extensions.screenshots.upload-disabled true
setpref extensions.webcompat-reporter.enabled false
setpref extensions.webextensions.restrictedDomains ""
setpref font.blacklist.underline_offset ""
setpref general.warnOnAboutConfig false
setpref geo.enabled false
setpref geo.wifi.uri ""
setpref intl.accept_languages "en-US, en"
setpref intl.locale.requested "en-US"
setpref intl.regional_prefs.use_os_locales false
setpref javascript.use_us_english_locale true
setpref layout.css.font-loading-api.enabled false
setpref media.autoplay.default 1
setpref media.eme.enabled false
setpref media.gmp-gmpopenh264.autoupdate false
setpref media.gmp-gmpopenh264.enabled false
setpref media.gmp-manager.updateEnabled false
setpref media.gmp-manager.url "data:text/plain,"
setpref media.gmp-manager.url.override "data:text/plain,"
setpref media.gmp-provider.enabled false
setpref media.gmp-widevinecdm.autoupdate false
setpref media.gmp-widevinecdm.enabled false
setpref media.gmp-widevinecdm.visible false
setpref media.gmp.trial-create.enabled false
" WebRTC. Might need to re-enable some day
setpref media.peerconnection.enabled false
setpref network.IDN_show_punycode true
setpref network.allow-experiments false
setpref network.http.referer.XOriginPolicy 1
setpref network.http.referer.defaultPolicy 3
setpref network.http.referer.defaultPolicy.pbmode 2
setpref network.http.referer.spoofSource false
setpref pdfjs.disabled true
setpref permissions.default.geo 0
setpref plugin.default.state 0
setpref plugin.defaultXpi.state 0
setpref plugin.sessionPermissionNow.intervalInMinutes 0
setpref plugins.click_to_play true
setpref privacy.firstparty.isolate true
" Disabled until https://bugzilla.mozilla.org/show_bug.cgi?id=1450398 is fixed
setpref privacy.resistFingerprinting false
setpref privacy.resistFingerprinting.block_mozAddonManager true
setpref privacy.userContext.enabled true
setpref privacy.userContext.ui.enabled true
setpref privacy.usercontext.about_newtab_segregation.enabled true
setpref reader.parse-on-load.enabled false
setpref security.dialog_enable_delay 500
setpref security.insecure_field_warning.contextual.enabled true
setpref signon.autofillForms false
setpref signon.rememberSignons false
setpref toolkit.cosmeticAnimations.enabled false
setpref toolkit.telemetry.archive.enabled false
setpref toolkit.telemetry.bhrPing.enabled false
setpref toolkit.telemetry.cachedClientID ""
setpref toolkit.telemetry.enabled false
setpref toolkit.telemetry.firstShutdownPing.enabled false
setpref toolkit.telemetry.hybridContent.enabled false
setpref toolkit.telemetry.newProfilePing.enabled false
setpref toolkit.telemetry.server "data:,"
setpref toolkit.telemetry.shutdownPingSender.enabled false
setpref toolkit.telemetry.unified false
setpref toolkit.telemetry.updatePing.enabled false
setpref ui.key.menuAccessKeyFocuses false
setpref xpinstall.signatures.required false

421
scripts/mutt_oauth2.py Executable file
View file

@ -0,0 +1,421 @@
#!/usr/bin/env python3
#
# Mutt OAuth2 token management script, version 2020-08-07
# Written against python 3.7.3, not tried with earlier python versions.
#
# Copyright (C) 2020 Alexander Perlis
#
# This program is free software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License as
# published by the Free Software Foundation; either version 2 of the
# License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
# General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
# 02110-1301, USA.
'''Mutt OAuth2 token management'''
import sys
import json
import argparse
import urllib.parse
import urllib.request
import imaplib
import poplib
import smtplib
import base64
import secrets
import hashlib
import time
from datetime import timedelta, datetime
from pathlib import Path
import socket
import http.server
import subprocess
import readline
# The token file must be encrypted because it contains multi-use bearer tokens
# whose usage does not require additional verification. Specify whichever
# encryption and decryption pipes you prefer. They should read from standard
# input and write to standard output. The example values here invoke GPG,
# although won't work until an appropriate identity appears in the first line.
ENCRYPTION_PIPE = ['gpg', '--encrypt', '--recipient', 'chris']
DECRYPTION_PIPE = ['gpg', '--decrypt']
registrations = {
'google': {
'authorize_endpoint': 'https://accounts.google.com/o/oauth2/auth',
'devicecode_endpoint': 'https://oauth2.googleapis.com/device/code',
'token_endpoint': 'https://accounts.google.com/o/oauth2/token',
'redirect_uri': 'urn:ietf:wg:oauth:2.0:oob',
'imap_endpoint': 'imap.gmail.com',
'pop_endpoint': 'pop.gmail.com',
'smtp_endpoint': 'smtp.gmail.com',
'sasl_method': 'OAUTHBEARER',
'scope': 'https://mail.google.com/',
'client_id': '',
'client_secret': '',
},
'microsoft': {
'authorize_endpoint': 'https://login.microsoftonline.com/common/oauth2/v2.0/authorize',
'devicecode_endpoint': 'https://login.microsoftonline.com/common/oauth2/v2.0/devicecode',
'token_endpoint': 'https://login.microsoftonline.com/common/oauth2/v2.0/token',
'redirect_uri': 'https://login.microsoftonline.com/common/oauth2/nativeclient',
'tenant': 'common',
'imap_endpoint': 'outlook.office365.com',
'pop_endpoint': 'outlook.office365.com',
'smtp_endpoint': 'smtp.office365.com',
'sasl_method': 'XOAUTH2',
'scope': ('offline_access https://outlook.office.com/IMAP.AccessAsUser.All '
'https://outlook.office.com/POP.AccessAsUser.All '
'https://outlook.office.com/SMTP.Send'),
'client_id': 'ca278be4-fe67-444c-9eae-d85dd486ae2c',
'client_secret': 'An78Q~xvPz45DFH~lS~tZm3iVj-ApjUJNb~iKc0R',
},
}
ap = argparse.ArgumentParser(epilog='''
This script obtains and prints a valid OAuth2 access token. State is maintained in an
encrypted TOKENFILE. Run with "--verbose --authorize" to get started or whenever all
tokens have expired, optionally with "--authflow" to override the default authorization
flow. To truly start over from scratch, first delete TOKENFILE. Use "--verbose --test"
to test the IMAP/POP/SMTP endpoints.
''')
ap.add_argument('-v', '--verbose', action='store_true', help='increase verbosity')
ap.add_argument('-d', '--debug', action='store_true', help='enable debug output')
ap.add_argument('tokenfile', help='persistent token storage')
ap.add_argument('-a', '--authorize', action='store_true', help='manually authorize new tokens')
ap.add_argument('--authflow', help='authcode | localhostauthcode | devicecode')
ap.add_argument('-t', '--test', action='store_true', help='test IMAP/POP/SMTP endpoints')
args = ap.parse_args()
token = {}
path = Path(args.tokenfile)
if path.exists():
if 0o777 & path.stat().st_mode != 0o600:
sys.exit('Token file has unsafe mode. Suggest deleting and starting over.')
try:
sub = subprocess.run(DECRYPTION_PIPE, check=True, input=path.read_bytes(),
capture_output=True)
token = json.loads(sub.stdout)
except subprocess.CalledProcessError:
sys.exit('Difficulty decrypting token file. Is your decryption agent primed for '
'non-interactive usage, or an appropriate environment variable such as '
'GPG_TTY set to allow interactive agent usage from inside a pipe?')
def writetokenfile():
'''Writes global token dictionary into token file.'''
if not path.exists():
path.touch(mode=0o600)
if 0o777 & path.stat().st_mode != 0o600:
sys.exit('Token file has unsafe mode. Suggest deleting and starting over.')
sub2 = subprocess.run(ENCRYPTION_PIPE, check=True, input=json.dumps(token).encode(),
capture_output=True)
path.write_bytes(sub2.stdout)
if args.debug:
print('Obtained from token file:', json.dumps(token))
if not token:
if not args.authorize:
sys.exit('You must run script with "--authorize" at least once.')
print('Available app and endpoint registrations:', *registrations)
token['registration'] = input('OAuth2 registration: ')
token['authflow'] = input('Preferred OAuth2 flow ("authcode" or "localhostauthcode" '
'or "devicecode"): ')
token['email'] = input('Account e-mail address: ')
token['access_token'] = ''
token['access_token_expiration'] = ''
token['refresh_token'] = ''
writetokenfile()
if token['registration'] not in registrations:
sys.exit(f'ERROR: Unknown registration "{token["registration"]}". Delete token file '
f'and start over.')
registration = registrations[token['registration']]
authflow = token['authflow']
if args.authflow:
authflow = args.authflow
baseparams = {'client_id': registration['client_id']}
# Microsoft uses 'tenant' but Google does not
if 'tenant' in registration:
baseparams['tenant'] = registration['tenant']
def access_token_valid():
'''Returns True when stored access token exists and is still valid at this time.'''
token_exp = token['access_token_expiration']
return token_exp and datetime.now() < datetime.fromisoformat(token_exp)
def update_tokens(r):
'''Takes a response dictionary, extracts tokens out of it, and updates token file.'''
token['access_token'] = r['access_token']
token['access_token_expiration'] = (datetime.now() +
timedelta(seconds=int(r['expires_in']))).isoformat()
if 'refresh_token' in r:
token['refresh_token'] = r['refresh_token']
writetokenfile()
if args.verbose:
print(f'NOTICE: Obtained new access token, expires {token["access_token_expiration"]}.')
if args.authorize:
p = baseparams.copy()
p['scope'] = registration['scope']
if authflow in ('authcode', 'localhostauthcode'):
verifier = secrets.token_urlsafe(90)
challenge = base64.urlsafe_b64encode(hashlib.sha256(verifier.encode()).digest())[:-1]
redirect_uri = registration['redirect_uri']
listen_port = 0
if authflow == 'localhostauthcode':
# Find an available port to listen on
s = socket.socket()
s.bind(('127.0.0.1', 0))
listen_port = s.getsockname()[1]
s.close()
redirect_uri = 'http://localhost:'+str(listen_port)+'/'
# Probably should edit the port number into the actual redirect URL.
p.update({'login_hint': token['email'],
'response_type': 'code',
'redirect_uri': redirect_uri,
'code_challenge': challenge,
'code_challenge_method': 'S256'})
print(registration["authorize_endpoint"] + '?' +
urllib.parse.urlencode(p, quote_via=urllib.parse.quote))
authcode = ''
if authflow == 'authcode':
authcode = input('Visit displayed URL to retrieve authorization code. Enter '
'code from server (might be in browser address bar): ')
else:
print('Visit displayed URL to authorize this application. Waiting...',
end='', flush=True)
class MyHandler(http.server.BaseHTTPRequestHandler):
'''Handles the browser query resulting from redirect to redirect_uri.'''
# pylint: disable=C0103
def do_HEAD(self):
'''Response to a HEAD requests.'''
self.send_response(200)
self.send_header('Content-type', 'text/html')
self.end_headers()
def do_GET(self):
'''For GET request, extract code parameter from URL.'''
# pylint: disable=W0603
global authcode
querystring = urllib.parse.urlparse(self.path).query
querydict = urllib.parse.parse_qs(querystring)
if 'code' in querydict:
authcode = querydict['code'][0]
self.do_HEAD()
self.wfile.write(b'<html><head><title>Authorizaton result</title></head>')
self.wfile.write(b'<body><p>Authorization redirect completed. You may '
b'close this window.</p></body></html>')
with http.server.HTTPServer(('127.0.0.1', listen_port), MyHandler) as httpd:
try:
httpd.handle_request()
except KeyboardInterrupt:
pass
if not authcode:
sys.exit('Did not obtain an authcode.')
for k in 'response_type', 'login_hint', 'code_challenge', 'code_challenge_method':
del p[k]
p.update({'grant_type': 'authorization_code',
'code': authcode,
'client_secret': registration['client_secret'],
'code_verifier': verifier})
print('Exchanging the authorization code for an access token')
try:
response = urllib.request.urlopen(registration['token_endpoint'],
urllib.parse.urlencode(p).encode())
except urllib.error.HTTPError as err:
print(err.code, err.reason)
response = err
response = response.read()
if args.debug:
print(response)
response = json.loads(response)
if 'error' in response:
print(response['error'])
if 'error_description' in response:
print(response['error_description'])
sys.exit(1)
elif authflow == 'devicecode':
try:
response = urllib.request.urlopen(registration['devicecode_endpoint'],
urllib.parse.urlencode(p).encode())
except urllib.error.HTTPError as err:
print(err.code, err.reason)
response = err
response = response.read()
if args.debug:
print(response)
response = json.loads(response)
if 'error' in response:
print(response['error'])
if 'error_description' in response:
print(response['error_description'])
sys.exit(1)
print(response['message'])
del p['scope']
p.update({'grant_type': 'urn:ietf:params:oauth:grant-type:device_code',
'client_secret': registration['client_secret'],
'device_code': response['device_code']})
interval = int(response['interval'])
print('Polling...', end='', flush=True)
while True:
time.sleep(interval)
print('.', end='', flush=True)
try:
response = urllib.request.urlopen(registration['token_endpoint'],
urllib.parse.urlencode(p).encode())
except urllib.error.HTTPError as err:
# Not actually always an error, might just mean "keep trying..."
response = err
response = response.read()
if args.debug:
print(response)
response = json.loads(response)
if 'error' not in response:
break
if response['error'] == 'authorization_declined':
print(' user declined authorization.')
sys.exit(1)
if response['error'] == 'expired_token':
print(' too much time has elapsed.')
sys.exit(1)
if response['error'] != 'authorization_pending':
print(response['error'])
if 'error_description' in response:
print(response['error_description'])
sys.exit(1)
print()
else:
sys.exit(f'ERROR: Unknown OAuth2 flow "{token["authflow"]}. Delete token file and '
f'start over.')
update_tokens(response)
if not access_token_valid():
if args.verbose:
print('NOTICE: Invalid or expired access token; using refresh token '
'to obtain new access token.')
if not token['refresh_token']:
sys.exit('ERROR: No refresh token. Run script with "--authorize".')
p = baseparams.copy()
p.update({'client_secret': registration['client_secret'],
'refresh_token': token['refresh_token'],
'grant_type': 'refresh_token'})
try:
response = urllib.request.urlopen(registration['token_endpoint'],
urllib.parse.urlencode(p).encode())
except urllib.error.HTTPError as err:
print(err.code, err.reason)
response = err
response = response.read()
if args.debug:
print(response)
response = json.loads(response)
if 'error' in response:
print(response['error'])
if 'error_description' in response:
print(response['error_description'])
print('Perhaps refresh token invalid. Try running once with "--authorize"')
sys.exit(1)
update_tokens(response)
if not access_token_valid():
sys.exit('ERROR: No valid access token. This should not be able to happen.')
if args.verbose:
print('Access Token: ', end='')
print(token['access_token'])
def build_sasl_string(user, host, port, bearer_token):
'''Build appropriate SASL string, which depends on cloud server's supported SASL method.'''
if registration['sasl_method'] == 'OAUTHBEARER':
return f'n,a={user},\1host={host}\1port={port}\1auth=Bearer {bearer_token}\1\1'
if registration['sasl_method'] == 'XOAUTH2':
return f'user={user}\1auth=Bearer {bearer_token}\1\1'
sys.exit(f'Unknown SASL method {registration["sasl_method"]}.')
if args.test:
errors = False
imap_conn = imaplib.IMAP4_SSL(registration['imap_endpoint'])
sasl_string = build_sasl_string(token['email'], registration['imap_endpoint'], 993,
token['access_token'])
if args.debug:
imap_conn.debug = 4
try:
imap_conn.authenticate(registration['sasl_method'], lambda _: sasl_string.encode())
# Microsoft has a bug wherein a mismatch between username and token can still report a
# successful login... (Try a consumer login with the token from a work/school account.)
# Fortunately subsequent commands fail with an error. Thus we follow AUTH with another
# IMAP command before reporting success.
imap_conn.list()
if args.verbose:
print('IMAP authentication succeeded')
except imaplib.IMAP4.error as e:
print('IMAP authentication FAILED (does your account allow IMAP?):', e)
errors = True
pop_conn = poplib.POP3_SSL(registration['pop_endpoint'])
sasl_string = build_sasl_string(token['email'], registration['pop_endpoint'], 995,
token['access_token'])
if args.debug:
pop_conn.set_debuglevel(2)
try:
# poplib doesn't have an auth command taking an authenticator object
# Microsoft requires a two-line SASL for POP
# pylint: disable=W0212
pop_conn._shortcmd('AUTH ' + registration['sasl_method'])
pop_conn._shortcmd(base64.standard_b64encode(sasl_string.encode()).decode())
if args.verbose:
print('POP authentication succeeded')
except poplib.error_proto as e:
print('POP authentication FAILED (does your account allow POP?):', e.args[0].decode())
errors = True
# SMTP_SSL would be simpler but Microsoft does not answer on port 465.
smtp_conn = smtplib.SMTP(registration['smtp_endpoint'], 587)
sasl_string = build_sasl_string(token['email'], registration['smtp_endpoint'], 587,
token['access_token'])
smtp_conn.ehlo('test')
smtp_conn.starttls()
smtp_conn.ehlo('test')
if args.debug:
smtp_conn.set_debuglevel(2)
try:
smtp_conn.auth(registration['sasl_method'], lambda _=None: sasl_string)
if args.verbose:
print('SMTP authentication succeeded')
except smtplib.SMTPAuthenticationError as e:
print('SMTP authentication FAILED:', e)
errors = True
if errors:
sys.exit(1)