# -*- coding: utf-8; -*-
################################################################################
#
# WuttaFarm --Web app to integrate with and extend farmOS
# Copyright © 2026 Lance Edgar
#
# This file is part of WuttaFarm.
#
# WuttaFarm 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 3 of the License, or (at your option) any later
# version.
#
# WuttaFarm 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
# WuttaFarm. If not, see <http://www.gnu.org/licenses/>.
#
################################################################################
"""
Auth views
"""
from httpx_auth import OAuth2AuthorizationCode
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
from wuttafarm.web.util import save_farmos_oauth2_token
[docs]
class AuthView(base.AuthView):
"""
Auth views
"""
[docs]
def authenticate_user(self, session, username, password):
"""
We override the default logic in the following way:
First the normal auth handler is invoked, for the actual
authentication.
If a user was authenticated via farmOS OAuth2, then a "special
OAuth2 value" will have been cached on the user object itself
(from within the auth handler logic). We look for that value
here, so we can stash it in the user's web session for sake of
future requests.
"""
if user := super().authenticate_user(session, username, password):
farmos = self.app.get_farmos_handler()
# legacy
if farmos.farmos_py_is_v1():
if "farmos_oauth2_token" in user.__dict__:
self.request.session["farmos.oauth2.token"] = user.__dict__.pop(
"farmos_oauth2_token"
)
return user
# modern
if "farmos_oauth2" in user.__dict__:
self.request.session["farmos_oauth2"] = user.__dict__.pop(
"farmos_oauth2"
)
return user
return None
# nb. this method only used for legacy farmOS v1.x
def get_farmos_oauth2_session(self):
farmos = self.app.get_farmos_handler()
return OAuth2Session(
client_id=farmos.get_oauth2_client_id(),
scope=farmos.get_oauth2_scope(),
redirect_uri=self.request.route_url("farmos_oauth_callback"),
)
[docs]
def farmos_oauth_login(self):
"""
View to initiate OAuth2 workflow.
"""
farmos = self.app.get_farmos_handler()
# legacy
if farmos.farmos_py_is_v1():
oauth2 = self.get_farmos_oauth2_session()
auth_url, state = oauth2.authorization_url(
farmos.get_farmos_url("/oauth/authorize")
)
return self.redirect(auth_url)
# modern
# TODO: the code below actually works fine..BUT unfortunately
# the httpx_auth logic for this workflow wants to spin up its
# own web service for the redirect, and then use python's
# webbrowser module to open a new client for that request...
# long story short, this seems to imply the 'authorization
# code' workflow is only meant for CLI/script usage? actually
# it worked "fine" for me in dev but broke when the app was
# deployed to production.
raise NotImplementedError(
"OAuth2 'authorization code' workflow "
"is not supported using farmOS.py v2.x"
)
# auth = self.app.get_auth_handler()
# session = Session()
# # nb. if login succeeds, we'll stash this in user session
# oauth2 = OAuth2AuthorizationCode(
# farmos.get_farmos_url("/oauth/authorize"),
# farmos.get_farmos_url("/oauth/token"),
# client_id=farmos.get_oauth2_client_id(),
# scope=farmos.get_oauth2_scope(),
# )
# # identify our user
# user = auth.identify_oauth2_user(oauth2, session)
# if not user:
# self.request.session.flash(
# "farmOS authentication was successful, but user is "
# f"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)
# self.request.session["farmos_oauth2"] = oauth2
# # prefer url from session over effective referrer
# referrer = self.request.get_referrer()
# referrer = self.request.session.pop("url_after_login", referrer)
# # but we don't want to redirect back to login!
# if referrer == self.request.route_url("login"):
# referrer = self.request.route_url("home")
# return self.redirect(referrer, headers=headers)
# nb. this view only used for legacy farmOS v1.x
[docs]
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
save_farmos_oauth2_token(self.request, token)
# nb. must give a *copy* of the token to farmOS client, since
# it will mutate it in-place and we don't want that to happen
# for our original copy in the user session. (otherwise the
# auto-refresh will not work correctly for subsequent calls.)
token = dict(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 "
f"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):
wutta_config = config.registry.settings.get("wutta_config")
app = wutta_config.get_app()
farmos = app.get_farmos_handler()
# 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
if farmos.farmos_py_is_v1():
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()
AuthView = kwargs.get("AuthView", local["AuthView"])
base.defaults(config, **{"AuthView": AuthView})
def includeme(config):
defaults(config)