diff --git a/pms_api_rest/services/pms_login_service.py b/pms_api_rest/services/pms_login_service.py index 690f9d9c..4530d802 100644 --- a/pms_api_rest/services/pms_login_service.py +++ b/pms_api_rest/services/pms_login_service.py @@ -43,7 +43,13 @@ def login(self, user): if not user_record: raise werkzeug.exceptions.Unauthorized(_("wrong user/pass")) try: - user_record.with_user(user_record)._check_credentials(user.password, None) + # This login cannot ask for a second factor, so it does not promise + # to: an account protected by one is refused here and has to use + # the login that can. An API key stays valid, which is the + # credential meant for callers with nobody in front of them. + user_record.with_user(user_record)._check_credentials( + user.password, {"interactive": False} + ) except AccessDenied as e: raise werkzeug.exceptions.Unauthorized(_("wrong user/pass")) from e diff --git a/pms_api_rest/tests/__init__.py b/pms_api_rest/tests/__init__.py index fb1a45be..c56521ed 100644 --- a/pms_api_rest/tests/__init__.py +++ b/pms_api_rest/tests/__init__.py @@ -8,3 +8,4 @@ from . import test_various_partner_protection from . import test_current_user_service from . import test_error_cors +from . import test_login_second_factor diff --git a/pms_api_rest/tests/test_login_second_factor.py b/pms_api_rest/tests/test_login_second_factor.py new file mode 100644 index 00000000..ba2a275b --- /dev/null +++ b/pms_api_rest/tests/test_login_second_factor.py @@ -0,0 +1,59 @@ +# Copyright 2026 Commit [Sun] +# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl). +"""This login cannot ask anyone for a verification code, so it must not let an +account that is protected by one through on its password alone. Those accounts +belong to the login that can ask, and everyone else keeps logging in here. +""" +import base64 +import os +from unittest.mock import patch + +import werkzeug.exceptions + +from odoo.tests import tagged + +from odoo.addons.base_rest.controllers.main import _PseudoCollection +from odoo.addons.component.core import WorkContext +from odoo.addons.pms.tests.common import TestPms + +IMAGE_URL_BUILDER = ( + "odoo.addons.pms_api_rest.services.pms_login_service.url_image_pms_api_rest" +) + + +@tagged("post_install", "-at_install") +class TestLoginSecondFactor(TestPms): + @classmethod + def setUpClass(cls): + super().setUpClass() + cls.api_user = cls.env["res.users"].create( + { + "name": "Api Caller", + "login": "api.caller@example.org", + "password": "supersecret", + "company_id": cls.company1.id, + "company_ids": [(6, 0, [cls.company1.id])], + "pms_property_ids": [(6, 0, [cls.pms_property1.id])], + "pms_property_id": cls.pms_property1.id, + } + ) + + def _login(self): + collection = _PseudoCollection("pms.services", self.env) + work = WorkContext( + model_name="rest.service.registration", collection=collection + ) + credentials = self.env.datamodels["pms.api.rest.user.input"]( + username="api.caller@example.org", password="supersecret" + ) + with patch(IMAGE_URL_BUILDER, return_value=""): + return work.component(usage="login").login(credentials) + + def test_an_account_without_a_second_factor_still_logs_in(self): + self.assertTrue(self._login().token) + + def test_an_account_with_a_second_factor_is_refused(self): + self.api_user.sudo().totp_secret = base64.b32encode(os.urandom(20)).decode() + + with self.assertRaises(werkzeug.exceptions.Unauthorized): + self._login() diff --git a/pms_fastapi/__manifest__.py b/pms_fastapi/__manifest__.py index bf59a1da..588ab03d 100644 --- a/pms_fastapi/__manifest__.py +++ b/pms_fastapi/__manifest__.py @@ -10,6 +10,7 @@ "fastapi_auth_jwt", "extendable_fastapi", "auth_jwt_login", + "auth_totp", "partner_firstname", "phone_validation", "account_payment_partner", diff --git a/pms_fastapi/models/__init__.py b/pms_fastapi/models/__init__.py index 113f693f..b4675749 100644 --- a/pms_fastapi/models/__init__.py +++ b/pms_fastapi/models/__init__.py @@ -11,3 +11,4 @@ from . import account_payment_method_line from . import pms_payment_refund_line from . import account_payment +from . import pms_mfa_challenge diff --git a/pms_fastapi/models/pms_mfa_challenge.py b/pms_fastapi/models/pms_mfa_challenge.py new file mode 100644 index 00000000..4c9cf0c4 --- /dev/null +++ b/pms_fastapi/models/pms_mfa_challenge.py @@ -0,0 +1,30 @@ +from odoo import api, fields, models + + +class PmsFastapiMfaChallenge(models.Model): + """A login that passed the password and still owes its second factor. + + It is the API's equivalent of the pre-authenticated session the web login + keeps: on its own it grants nothing, it only says which user is allowed to + hand in a code. Attempts are counted here so that a wrong code costs the + caller its challenge instead of its account. + """ + + _name = "pms.fastapi.mfa.challenge" + _description = "Pending second factor" + + user_id = fields.Many2one( + "res.users", required=True, index=True, ondelete="cascade" + ) + token = fields.Char(required=True, index=True) + expire = fields.Datetime(required=True) + attempts = fields.Integer(default=0) + + _sql_constraints = [ + ("unique_token", "unique(token)", "The token must be unique!"), + ] + + @api.autovacuum + def _remove_expired_challenges(self): + records = self.search([("expire", "<", fields.Datetime.now())]) + return records.unlink() diff --git a/pms_fastapi/routers/login.py b/pms_fastapi/routers/login.py index bd986310..f6afe683 100644 --- a/pms_fastapi/routers/login.py +++ b/pms_fastapi/routers/login.py @@ -1,41 +1,90 @@ -from fastapi import HTTPException, Response, status +import re +import secrets +from datetime import timedelta -from odoo import _, models +from fastapi import Request, Response, status +from fastapi.responses import JSONResponse + +from odoo import _, fields, models from odoo.exceptions import AccessDenied from odoo.addons.pms_fastapi.dependencies import PublicEnv from odoo.addons.pms_fastapi.models.fastapi_endpoint import pms_api_router -from odoo.addons.pms_fastapi.schemas.pms_login import PmsLoginInput +from odoo.addons.pms_fastapi.schemas.pms_login import PmsLoginInput, PmsMfaInput + +MFA_COOKIE = "mfa" +MFA_MAX_AGE = 5 * 60 +MFA_MAX_ATTEMPTS = 5 +DEVICE_COOKIE = "pms_td" +DEVICE_SCOPE = "api" +DEVICE_MAX_AGE = 90 * 86400 +# The only second factor this API knows how to ask for. +SUPPORTED_MFA_TYPE = "totp" + +UNAUTHORIZED_RESPONSE = { + "description": "Unauthorized", + "content": { + "application/problem+json": { + "example": { + "type": "/errors/invalid-credentials", + "title": "Invalid credentials", + "status": 401, + "detail": "Wrong user or password.", + } + } + }, +} @pms_api_router.post( "/login", status_code=204, - responses={ - 401: { - "description": "Unauthorized", - "content": {"application/json": {"example": {"detail": "wrong user/pass"}}}, - }, - 204: {"model": None}, - }, + responses={401: UNAUTHORIZED_RESPONSE, 204: {"model": None}}, tags=["login"], ) -async def login(user: PmsLoginInput, env: PublicEnv): +async def login(user: PmsLoginInput, request: Request, env: PublicEnv): """ - If the login is correct, sets the authorization cookies. + Starts a session. + + When the account is protected by a second factor the answer is + `/errors/mfa-required` and the code has to be sent to `/login/mfa`, unless + this device was remembered on a previous login. """ + endpoint = env["pms.fastapi.login.endpoint"] user_record = env["res.users"].sudo().search([("login", "=", user.username)]) - if not user_record: - env["pms.fastapi.login.endpoint"]._raise_wrong_credentials() + return endpoint._invalid_credentials_response() try: + # A person is waiting on the other side, so a password is the expected + # credential here; the second factor is asked for below. user_record.with_user(user_record)._check_credentials( - user.password.get_secret_value(), {"interactive": False} + user.password.get_secret_value(), {"interactive": True} ) except AccessDenied: - env["pms.fastapi.login.endpoint"]._raise_wrong_credentials() - return env["pms.fastapi.login.endpoint"]._get_login_response_with_cookies( - user_record + return endpoint._invalid_credentials_response() + return endpoint._authenticated_response( + user_record, request.cookies.get(DEVICE_COOKIE) + ) + + +@pms_api_router.post( + "/login/mfa", + status_code=204, + responses={401: UNAUTHORIZED_RESPONSE, 204: {"model": None}}, + tags=["login"], +) +async def login_mfa(verification: PmsMfaInput, request: Request, env: PublicEnv): + """ + Finishes a login that was answered with `/errors/mfa-required`. + + A wrong code can be tried again. Once the answer is + `/errors/mfa-too-many-attempts` the whole login has to start over. + """ + return env["pms.fastapi.login.endpoint"]._verification_response( + request.cookies.get(MFA_COOKIE), + verification.code, + verification.remember, + request.headers.get("user-agent"), ) @@ -60,10 +109,139 @@ class PmsFastapiLoginEndpoint(models.AbstractModel): _name = "pms.fastapi.login.endpoint" _description = "login endpoint helper" - def _raise_wrong_credentials(self): - raise HTTPException( - status_code=401, - detail=_("wrong user/pass"), + def _invalid_credentials_response(self): + return self._problem_response( + "/errors/invalid-credentials", + _("Invalid credentials"), + _("Wrong user or password."), + ) + + def _problem_response(self, type_, title, detail): + return JSONResponse( + status_code=status.HTTP_401_UNAUTHORIZED, + media_type="application/problem+json", + content={ + "type": type_, + "title": title, + "status": status.HTTP_401_UNAUTHORIZED, + "detail": detail, + }, + ) + + def _authenticated_response(self, user_record, device_token): + """Either a session, or the demand for a second factor.""" + mfa_type = user_record._mfa_type() + if not mfa_type: + return self._get_login_response_with_cookies(user_record) + if device_token and self._is_remembered_device(user_record, device_token): + return self._get_login_response_with_cookies(user_record) + if mfa_type != SUPPORTED_MFA_TYPE: + # Letting an account through because its second factor is one this + # API cannot ask for would turn the protection into a hole. + return self._problem_response( + "/errors/mfa-required", + _("Second factor required"), + _("This account cannot be used from here."), + ) + return self._demand_verification_response(user_record) + + def _demand_verification_response(self, user_record): + token = secrets.token_urlsafe(32) + self.env["pms.fastapi.mfa.challenge"].sudo().create( + { + "user_id": user_record.id, + "token": token, + "expire": fields.Datetime.now() + timedelta(seconds=MFA_MAX_AGE), + } + ) + response = self._problem_response( + "/errors/mfa-required", + _("Second factor required"), + _("Send the code shown by your authentication app."), + ) + self._set_cookie(response, MFA_COOKIE, token, MFA_MAX_AGE) + return response + + def _verification_response(self, token, code, remember, user_agent=None): + challenge = self._open_challenge(token) + if not challenge: + return self._start_over_response() + user_record = challenge.user_id + try: + with user_record._assert_can_auth(user=user_record.id): + user_record._totp_check(int(re.sub(r"\s", "", code))) + except (AccessDenied, ValueError): + # Returned and not raised: the attempt has to survive the answer, + # or a wrong code would cost the caller nothing. + challenge.attempts += 1 + if challenge.attempts >= MFA_MAX_ATTEMPTS: + challenge.unlink() + return self._start_over_response() + return self._problem_response( + "/errors/mfa-invalid-code", + _("Invalid code"), + _("The code is not valid."), + ) + challenge.unlink() + response = self._get_login_response_with_cookies(user_record) + if remember: + self._remember_device(user_record, response, user_agent) + self._set_cookie(response, MFA_COOKIE, "", 0) + return response + + def _start_over_response(self): + return self._problem_response( + "/errors/mfa-too-many-attempts", + _("Verification no longer possible"), + _("Log in again."), + ) + + def _open_challenge(self, token): + Challenge = self.env["pms.fastapi.mfa.challenge"].sudo() + if not token: + return Challenge + return Challenge.search( + [("token", "=", token), ("expire", ">", fields.Datetime.now())], limit=1 + ) + + def _is_remembered_device(self, user_record, token): + return ( + self.env["auth_totp.device"] + .sudo() + ._check_credentials_for_uid( + scope=DEVICE_SCOPE, key=token, uid=user_record.id + ) + ) + + def _remember_device(self, user_record, response, user_agent): + # Generated as the user: the key belongs to whoever is logging in, not + # to the account running the API. + key = ( + self.env["auth_totp.device"] + .with_user(user_record) + .sudo() + ._generate(DEVICE_SCOPE, self._device_name(user_agent)) + ) + self._set_cookie(response, DEVICE_COOKIE, key, DEVICE_MAX_AGE) + + def _device_name(self, user_agent): + """What the user will read in the trusted devices list.""" + return (user_agent or _("Unknown device"))[:128] + + def _set_cookie(self, response, key, value, max_age): + validator = ( + self.env["auth.jwt.validator"].sudo()._get_validator_by_name("api_pms") + ) + response.set_cookie( + key=key, + value=value, + max_age=max_age, + path=validator.cookie_path or "/", + secure=validator.cookie_secure, + httponly=True, + # The client always lives on a sibling host of this one, so the + # cookie is first party and does not need the cross-site opt-in. + samesite="Lax", ) def _get_logout_response(self): @@ -81,7 +259,7 @@ def _drop_session_cookies(self, response): path=validator.cookie_path or "/", secure=validator.cookie_secure, httponly=True, - samesite="None", + samesite="Lax", ) def _get_login_response_with_cookies(self, user_record): @@ -98,13 +276,7 @@ def _get_login_response_with_cookies(self, user_record): expire=validator.cookie_max_age, ) response = Response(status_code=status.HTTP_204_NO_CONTENT) - response.set_cookie( - key=validator.cookie_name, - value=token, - max_age=validator.cookie_max_age, - path=validator.cookie_path or "/", - secure=validator.cookie_secure, - httponly=True, - samesite="None", + self._set_cookie( + response, validator.cookie_name, token, validator.cookie_max_age ) return response diff --git a/pms_fastapi/schemas/pms_login.py b/pms_fastapi/schemas/pms_login.py index 6fe36ffc..b09efc8d 100644 --- a/pms_fastapi/schemas/pms_login.py +++ b/pms_fastapi/schemas/pms_login.py @@ -6,3 +6,8 @@ class PmsLoginInput(PmsBaseModel): username: str password: SecretStr + + +class PmsMfaInput(PmsBaseModel): + code: str + remember: bool = False diff --git a/pms_fastapi/security/ir.model.access.csv b/pms_fastapi/security/ir.model.access.csv index 2fda8bc0..21c62ac4 100644 --- a/pms_fastapi/security/ir.model.access.csv +++ b/pms_fastapi/security/ir.model.access.csv @@ -1,2 +1,3 @@ id,name,model_id:id,group_id:id,perm_read,perm_write,perm_create,perm_unlink access_pms_payment_refund_line_user,pms.payment.refund.line user,model_pms_payment_refund_line,base.group_user,1,1,1,1 +access_pms_fastapi_mfa_challenge_system,pms.fastapi.mfa.challenge system,model_pms_fastapi_mfa_challenge,base.group_system,1,1,1,1 diff --git a/pms_fastapi/tests/__init__.py b/pms_fastapi/tests/__init__.py index e95bbcfa..98ccd676 100644 --- a/pms_fastapi/tests/__init__.py +++ b/pms_fastapi/tests/__init__.py @@ -27,3 +27,4 @@ from . import test_search_text_guard from . import test_reservation_offboarding from . import test_logout +from . import test_login_second_factor diff --git a/pms_fastapi/tests/test_login_second_factor.py b/pms_fastapi/tests/test_login_second_factor.py new file mode 100644 index 00000000..5ad94373 --- /dev/null +++ b/pms_fastapi/tests/test_login_second_factor.py @@ -0,0 +1,120 @@ +import base64 +import os +import time + +from fastapi import status + +from odoo.addons.auth_totp.models.totp import hotp +from odoo.addons.pms_fastapi.tests.common import CommonTestPmsApi + +CREDENTIALS = {"username": "test_pms_api", "password": "supersecret"} + + +class TestLoginSecondFactor(CommonTestPmsApi): + @classmethod + def setUpClass(cls): + super().setUpClass() + cls.secret = base64.b32encode(os.urandom(20)).decode() + cls.test_user.sudo().totp_secret = cls.secret + + def _valid_code(self): + key = base64.b32decode(self.secret) + return str(hotp(key, int(time.time() / 30))) + + def _wrong_code(self): + return str((int(self._valid_code()) + 500000) % 1000000) + + def _problem(self, response): + self.assertEqual( + response.status_code, status.HTTP_401_UNAUTHORIZED, response.text + ) + return response.json()["type"] + + def test_password_alone_is_not_enough(self): + with self._create_test_client() as test_client: + response = test_client.post("/login", json=CREDENTIALS) + + self.assertEqual(self._problem(response), "/errors/mfa-required") + + def test_valid_code_completes_the_login(self): + with self._create_test_client() as test_client: + test_client.post("/login", json=CREDENTIALS) + + response = test_client.post( + "/login/mfa", json={"code": self._valid_code(), "remember": False} + ) + + self.assertEqual( + response.status_code, status.HTTP_204_NO_CONTENT, response.text + ) + authenticated = test_client.get("/agencies") + self.assertEqual(authenticated.status_code, status.HTTP_200_OK) + + def test_wrong_code_can_be_tried_again(self): + with self._create_test_client() as test_client: + test_client.post("/login", json=CREDENTIALS) + + response = test_client.post( + "/login/mfa", json={"code": self._wrong_code(), "remember": False} + ) + self.assertEqual(self._problem(response), "/errors/mfa-invalid-code") + + retried = test_client.post( + "/login/mfa", json={"code": self._valid_code(), "remember": False} + ) + self.assertEqual( + retried.status_code, status.HTTP_204_NO_CONTENT, retried.text + ) + + def test_enough_wrong_codes_end_the_login(self): + with self._create_test_client() as test_client: + test_client.post("/login", json=CREDENTIALS) + + for _attempt in range(4): + response = test_client.post( + "/login/mfa", json={"code": self._wrong_code(), "remember": False} + ) + self.assertEqual(self._problem(response), "/errors/mfa-invalid-code") + + response = test_client.post( + "/login/mfa", json={"code": self._wrong_code(), "remember": False} + ) + self.assertEqual(self._problem(response), "/errors/mfa-too-many-attempts") + + # The challenge is gone, so even the right code is of no use now. + response = test_client.post( + "/login/mfa", json={"code": self._valid_code(), "remember": False} + ) + self.assertEqual(self._problem(response), "/errors/mfa-too-many-attempts") + + def test_a_code_is_useless_without_its_login(self): + with self._create_test_client() as test_client: + response = test_client.post( + "/login/mfa", json={"code": self._valid_code(), "remember": False} + ) + + self.assertEqual(self._problem(response), "/errors/mfa-too-many-attempts") + + def test_a_remembered_device_is_not_asked_again(self): + with self._create_test_client() as test_client: + test_client.post("/login", json=CREDENTIALS) + test_client.post( + "/login/mfa", json={"code": self._valid_code(), "remember": True} + ) + + response = test_client.post("/login", json=CREDENTIALS) + + self.assertEqual( + response.status_code, status.HTTP_204_NO_CONTENT, response.text + ) + + def test_a_device_is_remembered_only_when_asked(self): + with self._create_test_client() as test_client: + test_client.post("/login", json=CREDENTIALS) + test_client.post( + "/login/mfa", json={"code": self._valid_code(), "remember": False} + ) + + response = test_client.post("/login", json=CREDENTIALS) + + self.assertEqual(self._problem(response), "/errors/mfa-required") diff --git a/roomdoo_fastapi/routers/login.py b/roomdoo_fastapi/routers/login.py index c3ef2f7b..a9b98c97 100644 --- a/roomdoo_fastapi/routers/login.py +++ b/roomdoo_fastapi/routers/login.py @@ -1,3 +1,5 @@ +import secrets + from fastapi import HTTPException, Request from odoo import models @@ -63,7 +65,10 @@ def _get_login_response_with_cookies(self, user_record): self.env["auth.jwt.validator"].sudo()._get_validator_by_name("api_pms") ) response = super()._get_login_response_with_cookies(user_record) - payload = {} + # Without something unique of its own the token is a pure function of + # its expiry, so two logins within the same second produce the very + # same string and the second one hits the uniqueness constraint. + payload = {"jti": secrets.token_urlsafe(16)} refresh_token = validator._encode( payload, expire=validator.refresh_cookie_max_age, @@ -77,7 +82,7 @@ def _get_login_response_with_cookies(self, user_record): path=validator.refresh_token_path or "/", secure=validator.cookie_secure, httponly=True, - samesite="None", + samesite="Lax", ) return response