fix: add support for farmOS/OAuth2 Authorization Code grant/workflow

This commit is contained in:
Lance Edgar 2026-02-06 14:24:41 -06:00
parent 039aa60038
commit 5b96fcfc2a
2 changed files with 119 additions and 0 deletions

View file

@ -0,0 +1,33 @@
## -*- coding: utf-8; -*-
<%inherit file="wuttaweb:templates/auth/login.mako" />
<%namespace name="base_meta" file="/base_meta.mako" />
<%def name="page_content()">
<div class="wutta-page-content">
<div class="wutta-logo">${base_meta.full_logo(image_url or None)}</div>
<div style="display: flex; gap: 1rem; align-items: center;">
<div class="card">
<div class="card-content">
${form.render_vue_tag()}
</div>
</div>
<div class="has-text-italic">
- OR -
</div>
<div class="card">
<div class="card-content">
<wutta-button type="is-primary"
tag="a"
href="${url('farmos_oauth_login')}"
icon-left="user"
label="Login via farmOS / OAuth2"
once />
</div>
</div>
</div>
</div>
</%def>

View file

@ -23,7 +23,12 @@
Auth views
"""
from oauthlib.oauth2 import AccessDeniedError
from requests_oauthlib import OAuth2Session
from wuttaweb.views import auth as base
from wuttaweb.auth import login_user
from wuttaweb.db import Session
class AuthView(base.AuthView):
@ -47,6 +52,87 @@ class AuthView(base.AuthView):
return None
def get_farmos_oauth2_session(self):
return OAuth2Session(
client_id="farm",
scope="farm_manager",
redirect_uri=self.request.route_url("farmos_oauth_callback"),
)
def farmos_oauth_login(self):
"""
View to initiate OAuth2 workflow.
"""
oauth = self.get_farmos_oauth2_session()
auth_url, state = oauth.authorization_url(
self.app.get_farmos_url("/oauth/authorize")
)
return self.redirect(auth_url)
def farmos_oauth_callback(self):
"""
View for OAuth2 workflow, provided as redirect URL when
authorizing.
"""
session = Session()
auth = self.app.get_auth_handler()
oauth = self.get_farmos_oauth2_session()
try:
# get oauth token from farmOS
token = oauth.fetch_token(
self.app.get_farmos_url("/oauth/token"),
authorization_response=self.request.current_route_url(),
include_client_id=True,
)
except AccessDeniedError:
self.request.session.flash("Access to farmOS was denied.", "error")
return self.redirect(self.request.route_url("login"))
# save token in user session
self.request.session["farmos.oauth2.token"] = token
# get (or create) native app user
farmos_client = self.app.get_farmos_client(token=token)
info = farmos_client.info()
farmos_user = farmos_client.resource.get_id(
"user", "user", info["meta"]["links"]["me"]["meta"]["id"]
)
user = auth.get_or_make_farmos_user(
session, farmos_user["data"]["attributes"]["name"]
)
if not user:
self.request.session.flash(
"farmOS authentication was successful, but user is "
"not allowed login to {self.app.get_node_title()}.",
"error",
)
return self.redirect(self.request.route_url("login"))
# delare user is logged in
headers = login_user(self.request, user)
referrer = self.request.get_referrer()
return self.redirect(referrer, headers=headers)
@classmethod
def defaults(cls, config):
cls._auth_defaults(config)
cls._wuttafarm_defaults(config)
@classmethod
def _wuttafarm_defaults(cls, config):
# farmos oauth login
config.add_route("farmos_oauth_login", "/farmos/oauth/login")
config.add_view(cls, attr="farmos_oauth_login", route_name="farmos_oauth_login")
# farmos oauth callback
config.add_route("farmos_oauth_callback", "/farmos/oauth/callback")
config.add_view(
cls, attr="farmos_oauth_callback", route_name="farmos_oauth_callback"
)
def defaults(config, **kwargs):
local = globals()