Source code for wuttafarm.web.views.auth

# -*- 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 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




class AuthView(base.AuthView):
    """
    Auth views
    """

    def authenticate_user(self, session, username, password):
        if user := super().authenticate_user(session, username, password):

            # nb. auth handler will stash the farmOS oauth2 token in
            # the user object, if applicable (i.e. unless the user
            # authenticated normally via the app DB).  when a token is
            # involved we need to put that in the user's web session.
            if "farmos_oauth2_token" in user.__dict__:
                self.request.session["farmos.oauth2.token"] = user.__dict__.pop(
                    "farmos_oauth2_token"
                )

            return user

        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
        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 "
                "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()
    AuthView = kwargs.get("AuthView", local["AuthView"])
    base.defaults(config, **{"AuthView": AuthView})


def includeme(config):
    defaults(config)