rattail-project/rattail Changeset

Changeset - 2b7ac6a5fc7b

Parent rev.

Child rev.

[Not reviewed]

0 5 1

Lance Edgar (lance) - 17 months ago 2023-05-15 08:07:33
lance@edbob.org

Add schema, basic logic for user API tokens

and the default API client now tries to use token if configured, or
can fallback to login w/ credentials

6 files changed with 180 insertions and 7 deletions:

rattail/auth.py

rattail/db/alembic/versions/4747a017b8f9_add_table_for_api_tokens.py

rattail/db/model/__init__.py

rattail/db/model/users.py

rattail/tailbone.py

rattail/telemetry.py

0 comments (0 inline, 0 general)

rattail/auth.py

➞

Show inline comments

@@ @@ -2,7 +2,7 @@ @@
 ################################################################################
+#
 #  Rattail -- Retail Software Framework
-#  Copyright © 2010-2022 Lance Edgar
+#  Copyright © 2010-2023 Lance Edgar
+#
 #  This file is part of Rattail.
+#
@@ @@ -26,10 +26,10 @@ Auth Handler @@
 See also :doc:`rattail-manual:base/handlers/other/auth`.
 """
 from __future__ import unicode_literals, absolute_import
 import secrets
 import warnings
 from sqlalchemy import orm
 import sqlalchemy_continuum as continuum
 from rattail.app import GenericHandler, MergeMixin
@@ @@ -77,6 +77,24 @@ class AuthHandler(GenericHandler, MergeMixin): @@
         return authenticate_user(session, username, password)
     def authenticate_user_token(self, session, token):
         """
         Authenticate the given user API token string, and if valid,
         return the corresponding User object.
         """
         model = self.model
         try:
             token = session.query(model.UserAPIToken)\
                            .filter(model.UserAPIToken.token_string == token)\
                            .one()
         except orm.exc.NoResultFound:
             pass
         else:
             user = token.user
             if user.active:
                 return user
     def has_permission(self, session, principal, permission,
                        include_guest=True,
                        include_authenticated=True):
@@ @@ -339,6 +357,39 @@ class AuthHandler(GenericHandler, MergeMixin): @@
         # TODO: remove this, legacy code
         return user.get_email_address()
     def generate_raw_api_token(self):
         """
         Generate a new *raw* API token string.
         """
         return secrets.token_urlsafe()
     def add_api_token(self, user, description, **kwargs):
         """
         Add a new API token for the user.
         """
         model = self.model
         session = self.app.get_session(user)
         # generate raw API token, in the form required for use within
         # the API client
         token_string = self.generate_raw_api_token()
         # create DB record for the token
         token = model.UserAPIToken(
             user=user,
             description=description,
             token_string=token_string)
         session.add(token)
         return token
     def delete_api_token(self, token, **kwargs):
         """
         Delete a new API token for the user.
         """
         session = self.app.get_session(token)
         session.delete(token)
     def get_merge_preview_fields(self, **kwargs):
         """
         Returns a sequence of fields which will be used during a merge

rattail/db/alembic/versions/4747a017b8f9_add_table_for_api_tokens.py

➞

Show inline comments

@@ new file 100644 @@
 # -*- coding: utf-8; -*-
 """add table for API Tokens
 Revision ID: 4747a017b8f9
 Revises: 0aeaac17cd6e
 Create Date: 2023-05-14 11:18:12.265992
 """
 # revision identifiers, used by Alembic.
 revision = '4747a017b8f9'
 down_revision = '0aeaac17cd6e'
 branch_labels = None
 depends_on = None
 from alembic import op
 import sqlalchemy as sa
 import rattail.db.types
 def upgrade():
     # user_api_token
     op.create_table('user_api_token',
                     sa.Column('uuid', sa.String(length=32), nullable=False),
                     sa.Column('user_uuid', sa.String(length=32), nullable=False),
                     sa.Column('description', sa.String(length=255), nullable=False),
                     sa.Column('token_string', sa.String(length=255), nullable=False),
                     sa.Column('created', sa.DateTime(), nullable=False),
                     sa.ForeignKeyConstraint(['user_uuid'], ['user.uuid'], name='user_api_token_fk_user'),
                     sa.PrimaryKeyConstraint('uuid')
+    )
     op.create_table('user_api_token_version',
                     sa.Column('uuid', sa.String(length=32), autoincrement=False, nullable=False),
                     sa.Column('user_uuid', sa.String(length=32), autoincrement=False, nullable=True),
                     sa.Column('description', sa.String(length=255), autoincrement=False, nullable=True),
                     sa.Column('token_string', sa.String(length=255), autoincrement=False, nullable=True),
                     sa.Column('created', sa.DateTime(), nullable=True),
                     sa.Column('transaction_id', sa.BigInteger(), autoincrement=False, nullable=False),
                     sa.Column('end_transaction_id', sa.BigInteger(), nullable=True),
                     sa.Column('operation_type', sa.SmallInteger(), nullable=False),
                     sa.PrimaryKeyConstraint('uuid', 'transaction_id')
+    )
     op.create_index(op.f('ix_user_api_token_version_end_transaction_id'), 'user_api_token_version', ['end_transaction_id'], unique=False)
     op.create_index(op.f('ix_user_api_token_version_operation_type'), 'user_api_token_version', ['operation_type'], unique=False)
     op.create_index(op.f('ix_user_api_token_version_transaction_id'), 'user_api_token_version', ['transaction_id'], unique=False)
 def downgrade():
     # user_api_token
     op.drop_index(op.f('ix_user_api_token_version_transaction_id'), table_name='user_api_token_version')
     op.drop_index(op.f('ix_user_api_token_version_operation_type'), table_name='user_api_token_version')
     op.drop_index(op.f('ix_user_api_token_version_end_transaction_id'), table_name='user_api_token_version')
     op.drop_table('user_api_token_version')
     op.drop_table('user_api_token')

rattail/db/model/__init__.py

➞

Show inline comments

 from .people import (Person, PersonNote,
                      PersonPhoneNumber, PersonEmailAddress, PersonMailingAddress,
                      MergePeopleRequest)
 from .users import Role, Permission, User, UserRole, UserEvent
+from .users import Role, Permission, User, UserRole, UserEvent, UserAPIToken
 from .stores import Store, StorePhoneNumber, StoreEmailAddress
 from .customers import (Customer,
                         CustomerPhoneNumber,

rattail/db/model/users.py

➞

Show inline comments

@@ @@ -366,3 +366,50 @@ class UserEvent(Base): @@
     occurred = sa.Column(sa.DateTime(), nullable=True, default=datetime.datetime.utcnow, doc="""
     Timestamp at which the event occurred.
     """)
 class UserAPIToken(Base):
     """
     User authentication token for use with Tailbone API
     """
     __tablename__ = 'user_api_token'
     __table_args__ = (
         sa.ForeignKeyConstraint(['user_uuid'], ['user.uuid'],
                                 name='user_api_token_fk_user'),
+    )
     __versioned__ = {}
     model_title = "API Token"
     model_title_plural = "API Tokens"
     uuid = uuid_column()
     user_uuid = sa.Column(sa.String(length=32), nullable=False, doc="""
     Reference to the User associated with the token.
     """)
     user = orm.relationship(
         'User',
         doc="""
         Reference to the User associated with the token.
         """,
         backref=orm.backref(
             'api_tokens',
             cascade_backrefs=False,
             order_by='UserAPIToken.created',
             doc="""
             List of API tokens for the user.
             """))
     description = sa.Column(sa.String(length=255), nullable=False, doc="""
     Description of the token.
     """)
     token_string = sa.Column(sa.String(length=255), nullable=False, doc="""
     Token string, to be used by API clients.
     """)
     created = sa.Column(sa.DateTime(), nullable=False, default=datetime.datetime.utcnow, doc="""
     Date/time when the token was created.
     """)
     def __str__(self):
         return self.description or ""

rattail/tailbone.py

➞

Show inline comments

@@ @@ -82,6 +82,12 @@ class TailboneAPIClient(object): @@
         self.session = requests.Session()
         # maybe *disable* SSL cert verification
         # (should only be used for testing! e.g. w/ self-signed certs)
         if not self.config.getbool('tailbone.api', 'ssl_verify', default=True):
             self.session.verify = False
         # maybe set max retries, e.g. for flaky connections
         if self.max_retries is not None:
             adapter = requests.adapters.HTTPAdapter(max_retries=self.max_retries)
             self.session.mount(self.base_url, adapter)
@@ @@ -91,12 +97,25 @@ class TailboneAPIClient(object): @@
         # 400 Client Error: Bad CSRF Origin for url
         parts = urlparse(self.base_url)
         self.session.headers.update({
             'Origin': '{}://{}'.format(parts.scheme, parts.netloc)})
             'Origin': f'{parts.scheme}://{parts.netloc}',
         })
         # fetch basic 'session' endpoint, to get current xsrf token
         # (this does not require any authentication, which is next)
         response = self.get('/session')
         self.session.headers.update({
             'X-XSRF-TOKEN': response.cookies['XSRF-TOKEN']})
         # authenticate via token (preferred), or user/pass login
         token = self.config.get('tailbone.api', 'token')
         if token:
             self.session.headers.update({
                 'Authorization': 'Bearer {}'.format(token),
             })
         else: # no token, so attempt login w/ credentials
             if not self.login():
                 raise RuntimeError("login failed! (consider using token auth)")
     def _request(self, request_method, api_method, params=None, data=None):
         """
         Perform a request for the given API method, and return the response.
@@ @@ -118,6 +137,7 @@ class TailboneAPIClient(object): @@
         """
         Perform a GET request for the given API method, and return the response.
         """
         self._init()
         return self._request('GET', api_method, params=params)
     def post(self, api_method, **kwargs):

rattail/telemetry.py

➞

Show inline comments

@@ @@ -333,8 +333,6 @@ class TelemetryHandler(GenericHandler): @@
     def submit_data_tailbone_api(self, data, profile, **kwargs):
         api = TailboneAPIClient(self.config)
         if not api.login():
             raise RuntimeError("login failed!")
         data['uuid'] = profile.submit_uuid
         api.post(profile.submit_url, data=data)

0 comments (0 inline, 0 general)