audrius/pms
1
0
Fork 1
mirror of https://github.com/OCA/pms.git synced 2025-01-29 00:17:45 +02:00
Code Issues Packages Projects Releases Wiki Activity

[IMP] pms-api-rest: adapt jwt auth to oca/rest-framework's lasts udpates

Browse Source
This commit is contained in:
miguelpadin
2021-12-26 12:06:28 +01:00
committed by Darío Lodeiros
parent 72d7600755
commit 7f6faae55d
13 changed files with 17 additions and 527 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
pms_api_rest/__init__.py
View File

@@ -1,4 +1,3 @@
from . import controllers from . import controllers
from . import datamodels from . import datamodels
from . import models
from . import services from . import services

3
pms_api_rest/__manifest__.py
View File

@@ -11,12 +11,13 @@
"base_rest_datamodel", "base_rest_datamodel",
"web", "web",
"auth_signup", "auth_signup",
"auth_jwt_login",
], ],
"external_dependencies": { "external_dependencies": {
"python": ["jwt", "simplejson", "marshmallow"], "python": ["jwt", "simplejson", "marshmallow"],
}, },
"data": [ "data": [
"security/ir.model.access.csv", "security/ir.model.access.csv", "data/auth_jwt_validator.xml"
], ],
"installable": True, "installable": True,
} }

1
pms_api_rest/controllers/__init__.py
View File

@@ -1,2 +1 @@
from . import jwt_controller
from . import pms_rest from . import pms_rest

110
pms_api_rest/controllers/jwt_controller.py
View File

@@ -1,110 +0,0 @@
import logging
import werkzeug
from odoo import http
from odoo.http import request
from odoo.addons.auth_signup.models.res_users import SignupError
from ..lib_jwt.jwt_http import jwt_http
from ..lib_jwt.validator import validator
_logger = logging.getLogger(__name__)
SENSITIVE_FIELDS = [
"password",
"password_crypt",
"new_password",
"create_uid",
"write_uid",
]
class JwtController(http.Controller):
# test route
@http.route("/api/info", auth="public", csrf=False, cors="*")
def index(self, **kw):
return "Hello, world"
@http.route(
"/api/login", type="http", auth="public", csrf=False, cors="*", methods=["POST"]
)
def login(self, email, password, **kw):
return jwt_http.do_login(email, password)
@http.route("/api/me", type="http", auth="public", csrf=False, cors="*")
def me(self, **kw):
http_method, body, headers, token = jwt_http.parse_request()
result = validator.verify_token(token)
if not result["status"]:
return jwt_http.errcode(code=result["code"], message=result["message"])
return jwt_http.response(request.env.user.to_dict(True))
@http.route("/api/logout", type="http", auth="public", csrf=False, cors="*")
def logout(self, **kw):
http_method, body, headers, token = jwt_http.parse_request()
result = validator.verify_token(token)
if not result["status"]:
return jwt_http.errcode(code=result["code"], message=result["message"])
jwt_http.do_logout(token)
return jwt_http.response()
@http.route(
"/api/register",
type="http",
auth="public",
csrf=False,
cors="*",
methods=["POST"],
)
def register(self, email=None, name=None, password=None, **kw):
if not validator.is_valid_email(email):
return jwt_http.errcode(code=400, message="Invalid email address")
if not name:
return jwt_http.errcode(code=400, message="Name cannot be empty")
if not password:
return jwt_http.errcode(code=400, message="Password cannot be empty")
# sign up
try:
self._signup_with_values(login=email, name=name, password=password)
except AttributeError:
return jwt_http.errcode(code=501, message="Signup is disabled")
except (SignupError, AssertionError) as e:
if request.env["res.users"].sudo().search([("login", "=", email)]):
return jwt_http.errcode(
code=400, message="Email address already exists"
)
else:
_logger.error("%s", e)
return jwt_http.response_500()
except Exception as e:
_logger.error(str(e))
return jwt_http.response_500()
# log the user in
return jwt_http.do_login(email, password)
def _signup_with_values(self, **values):
request.env["res.users"].sudo().signup(values, None)
request.env.cr.commit() # as authenticate will use its
# own cursor we need to commit the current transaction
self.signup_email(values)
def signup_email(self, values):
user_sudo = (
request.env["res.users"]
.sudo()
.search([("login", "=", values.get("login"))])
)
template = request.env.ref(
"auth_signup.mail_template_user_signup_account_created",
raise_if_not_found=False,
)
if user_sudo and template:
template.sudo().with_context(
lang=user_sudo.lang,
auth_login=werkzeug.url_encode({"auth_login": user_sudo.email}),
).send_mail(user_sudo.id, force_send=True)

17
pms_api_rest/controllers/pms_rest.py
View File

@@ -1,22 +1,7 @@
from odoo.addons.base_rest.controllers import main from odoo.addons.base_rest.controllers import main
from ..lib_jwt.jwt_http import jwt_http
from ..lib_jwt.validator import validator
class BaseRestDemoPublicApiController(main.RestController): class BaseRestDemoPublicApiController(main.RestController):
_root_path = "/api/" _root_path = "/api/"
_collection_name = "pms.reservation.service" _collection_name = "pms.reservation.service"
_default_auth = "public" _default_auth = "jwt_api_pms"
# RestController OVERRIDE method
def _process_method(self, service_name, method_name, *args, params=None):
http_method, body, headers, token = jwt_http.parse_request()
result = validator.verify_token(token)
if not result["status"]:
return jwt_http.errcode(code=result["code"], message=result["message"])
else:
return super(BaseRestDemoPublicApiController, self)._process_method(
service_name, method_name, *args, params=params
)

14
pms_api_rest/data/auth_jwt_validator.xml Normal file
View File

@@ -0,0 +1,14 @@
<odoo>
<record id="demo_validator" model="auth.jwt.validator">
<field name="name">api_pms</field>
<field name="audience">api_pms</field>
<field name="issuer">pms</field>
<field name="signature_type">secret</field>
<field name="secret_algorithm">HS256</field>
<field name="secret_key">pms_secret_key_example</field>
<field name="user_id_strategy">login</field>
<field name="static_user_id">1</field>
<field name="partner_id_strategy">email</field>
<field name="partner_id_required" eval="False" />
</record>
</odoo>

131
pms_api_rest/lib_jwt/jwt_http.py
View File

@@ -1,131 +0,0 @@
import datetime
import simplejson as json
from odoo import http
from odoo.exceptions import AccessDenied
from odoo.http import Response, request
from .validator import validator
return_fields = ["id", "login", "name", "company_id"]
class JwtHttp:
def get_state(self):
return {"d": request.session.db}
def parse_request(self):
http_method = request.httprequest.method
try:
body = http.request.params
except Exception:
body = {}
headers = dict(list(request.httprequest.headers.items()))
if "wsgi.input" in headers:
del headers["wsgi.input"]
if "wsgi.errors" in headers:
del headers["wsgi.errors"]
if "HTTP_AUTHORIZATION" in headers:
headers["Authorization"] = headers["HTTP_AUTHORIZATION"]
# extract token
token = ""
if "Authorization" in headers:
try:
# Bearer token_string
token = headers["Authorization"].split(" ")[1]
except Exception:
pass
return http_method, body, headers, token
def date2str(self, d, f="%Y-%m-%d %H:%M:%S"):
"""
Convert datetime to string
:param self:
:param d: datetime object
:param f='%Y-%m-%d%H:%M:%S': string format
"""
try:
s = d.strftime(f)
except Exception:
s = None
return s
def response(self, success=True, message=None, data=None, code=200):
"""
Create a HTTP Response for controller
:param success=True indicate this response is successful or not
:param message=None message string
:param data=None data to return
:param code=200 http status code
"""
payload = json.dumps(
{
"success": success,
"message": message,
"data": data,
}
)
return Response(
payload,
status=code,
headers=[
("Content-Type", "application/json"),
],
)
def response_500(self, message="Internal Server Error", data=None):
return self.response(success=False, message=message, data=data, code=500)
def response_401(self, message="401 Unauthorized", data=None):
return self.response(success=False, message=message, data=data, code=401)
def response_404(self, message="404 Not Found", data=None):
return self.response(success=False, message=message, data=data, code=404)
def response_403(self, message="403 Forbidden", data=None):
return self.response(success=False, message=message, data=data, code=403)
def errcode(self, code, message=None):
return self.response(success=False, code=code, message=message)
def do_login(self, login, password):
# get current db
state = self.get_state()
try:
uid = request.session.authenticate(state["d"], login, password)
except AccessDenied:
return self.response_401()
if not uid:
return self.response_401()
# login success, generate token
user = request.env.user.read(return_fields)[0]
exp = datetime.datetime.utcnow() + datetime.timedelta(minutes=30000)
token = validator.create_token(user, exp)
return self.response(
data={"user": user, "exp": json.dumps(exp.isoformat()), "token": token}
)
def do_logout(self, token):
request.session.logout()
request.env["jwt_provider.access_token"].sudo().search(
[("token", "=", token)]
).unlink()
return self.response()
def cleanup(self):
# Clean up things after success request
# use logout here to make request as stateless as possible
request.session.logout()
return self.response()
jwt_http = JwtHttp()

85
pms_api_rest/lib_jwt/util.py
View File

@@ -1,85 +0,0 @@
import logging
import os
import random
import string
from dateutil.parser import parse
_logger = logging.getLogger(__name__)
class Util:
addons_path = os.path.join(os.path.dirname(os.path.abspath(__file__)))
def __init__(self):
self.addons_path = self.addons_path.replace("jwt_provider", "")
def generate_verification_code(self, length=8):
return "".join(
random.choice(string.ascii_uppercase + string.digits) for _ in range(length)
)
def toDate(self, pgTimeStr):
return parse(pgTimeStr)
def path(self, *paths):
"""Make a path"""
return os.path.join(self.addons_path, *paths)
def add_branch(self, tree, vector, value):
"""
Given a dict, a vector, and a value, insert the value into the dict
at the tree leaf specified by the vector. Recursive!
Params:
data (dict): The data structure to insert the vector into.
vector (list): A list of values representing the path to the leaf node.
value (object): The object to be inserted at the leaf
Example 1:
tree = {'a': 'apple'}
vector = ['b', 'c', 'd']
value = 'dog'
tree = add_branch(tree, vector, value)
Returns:
tree = { 'a': 'apple', 'b': { 'c': {'d': 'dog'}}}
Example 2:
vector2 = ['b', 'c', 'e']
value2 = 'egg'
tree = add_branch(tree, vector2, value2)
Returns:
tree = { 'a': 'apple', 'b': { 'c': {'d': 'dog', 'e': 'egg'}}}
Returns:
dict: The dict with the value placed at the path specified.
Algorithm:
If we're at the leaf, add it as key/value to the tree
Else: If the subtree doesn't exist, create it.
Recurse with the subtree and the left shifted vector.
Return the tree.
"""
key = vector[0]
tree[key] = (
value
if len(vector) == 1
else self.add_branch(tree[key] if key in tree else {}, vector[1:], value)
)
return tree
def create_dict(self, d):
res = {}
for k, v in d.items():
ar = k.split(".")
filter(None, ar)
self.add_branch(res, ar, v)
return res
util = Util()

107
pms_api_rest/lib_jwt/validator.py
View File

@@ -1,107 +0,0 @@
import datetime
import logging
import re
import traceback
import jwt
from jwt import InvalidSignatureError
from odoo.http import request
from odoo.tools import DEFAULT_SERVER_DATETIME_FORMAT
_logger = logging.getLogger(__name__)
regex = (
r"^[a-z0-9!#$%&'*+\/=?^_`{|}~-]+(?:\.[a-z0-9!#$%&'*+\/=?^_`{|}~-]+)*@(?:[a-z0-9]"
)
regex += r"(?:[a-z0-9-]*[a-z0-9])?\.)+[a-z0-9](?:[a-z0-9-]*[a-z0-9])?$"
class Validator:
def is_valid_email(self, email):
return re.search(regex, email)
def key(self):
# TODO: change this key before production (build an UI Form to maintain
# (in form company?)
return "CHANGE THIS KEY"
def create_token(self, user, exp):
try:
payload = {
"exp": exp,
"iat": datetime.datetime.utcnow(),
"sub": user["id"],
"lgn": user["login"],
}
token = jwt.encode(payload, self.key(), algorithm="HS256")
self.save_token(token, user["id"], exp)
return token
except Exception as ex:
_logger.error(ex)
raise
def save_token(self, token, uid, exp):
request.env["jwt_provider.access_token"].sudo().create(
{
"user_id": uid,
"expires": exp.strftime(DEFAULT_SERVER_DATETIME_FORMAT),
"token": token,
}
)
def verify(self, token):
record = (
request.env["jwt_provider.access_token"]
.sudo()
.search([("token", "=", token)])
)
if len(record) != 1:
_logger.info("not found %s" % token)
return False
if record.is_expired:
return False
return record.user_id
def verify_token(self, token):
try:
result = {
"status": False,
"message": None,
}
if not self.verify(token):
result["message"] = "Token invalid or expired"
result["code"] = 498
_logger.info("11111")
return result
payload = jwt.decode(token, self.key(), algorithms=["HS256"])
uid = request.session.authenticate(
request.session.db, login=payload["lgn"], password=token
)
if not uid:
result["message"] = "Token invalid or expired"
result["code"] = 498
_logger.info("2222")
return result
result["status"] = True
return result
except (
jwt.ExpiredSignatureError,
jwt.InvalidTokenError,
InvalidSignatureError,
Exception,
):
result["code"] = 498
result["message"] = "Token invalid or expired"
_logger.error(traceback.format_exc())
return result
validator = Validator()

2
pms_api_rest/models/__init__.py
View File

@@ -1,2 +0,0 @@
from . import res_users
from . import jwt_access_token

29
pms_api_rest/models/jwt_access_token.py
View File

@@ -1,29 +0,0 @@
from datetime import datetime
from odoo import api, fields, models
class JwtAccessToken(models.Model):
_name = "jwt_provider.access_token"
_description = "Store user access token for one-time-login"
token = fields.Char("Access Token", required=True)
user_id = fields.Many2one(
comodel_name="res.users",
string="User",
required=True,
ondelete="cascade",
)
expires = fields.Datetime(
"Expires",
required=True,
)
is_expired = fields.Boolean(
compute="_compute_is_expired",
)
@api.depends("expires")
def _compute_is_expired(self):
for token in self:
token.is_expired = datetime.now() > token.expires

43
pms_api_rest/models/res_users.py
View File

@@ -1,43 +0,0 @@
import logging
from odoo import api, fields, models
from odoo.exceptions import AccessDenied
from ..lib_jwt.validator import validator
_logger = logging.getLogger(__name__)
class ResUsers(models.Model):
_inherit = "res.users"
access_token_ids = fields.One2many(
string="Access Tokens",
comodel_name="jwt_provider.access_token",
inverse_name="user_id",
)
@classmethod
def _login(cls, db, login, password, user_agent_env):
user_id = super(ResUsers, cls)._login(db, login, password, user_agent_env)
if user_id:
return user_id
uid = validator.verify(password)
_logger.info(uid)
return uid
@api.model
def _check_credentials(self, password, user_agent_env):
try:
super(ResUsers, self)._check_credentials(password, user_agent_env)
except AccessDenied:
if not validator.verify(password):
raise
def to_dict(self, single=False):
res = []
for u in self:
d = u.read(["email", "name", "company_id"])[0]
res.append(d)
return res[0] if single else res

1
pms_api_rest/services/partner_services.py
View File

@@ -19,7 +19,6 @@ class PmsPartnerService(Component):
) )
], ],
output_param=Datamodel("pms.partner.info", is_list=True), output_param=Datamodel("pms.partner.info", is_list=True),
auth="public",
) )
def get_partners(self): def get_partners(self):
domain = [] domain = []