fix test

Author: bruntib

web/client/codechecker_client/client.py

@@ -77,16 +77,17 @@ def login_user(protocol, host, port, username, login=False):
     If login is False the user will be logged out.
     """
     session = UserCredentials()
-    auth_client = ThriftAuthHelper(protocol, host, port,
-                                   '/v' + CLIENT_API + '/Authentication')
 
     if not login:
+        auth_client = init_auth_client(protocol, host, port)
         logout_done = auth_client.destroySession()
         if logout_done:
             session.save_token(host, port, None, True)
             LOG.info("Successfully logged out.")
         return
 
+    auth_client = setup_auth_client(protocol, host, port)
+
     try:
         handshake = auth_client.getAuthParameters()
 

web/server/codechecker_server/api/authentication.py

@@ -751,7 +751,8 @@ def newPersonalAccessToken(self, name, description):
             token = generate_session_token()
             user = self.getLoggedInUser()
             access_token = PersonalAccessTokenDB(
-                user, name, token, description, auth_session.id)
+                user, name, token, description,
+                auth_session.groups, auth_session.id)
 
             session.add(access_token)
 
@@ -780,10 +781,10 @@ def removePersonalAccessToken(self, name):
         self.__require_privilaged_access()
         with DBSession(self.__config_db) as session:
             user = self.getLoggedInUser()
-            personal_access_token = session.query(PersonalAccessTokenDB) \
+            personal_access_token_q = session.query(PersonalAccessTokenDB) \
                 .filter(PersonalAccessTokenDB.user_name == user) \
-                .filter(PersonalAccessTokenDB.token_name == name) \
-                .first()
+                .filter(PersonalAccessTokenDB.token_name == name)
+            personal_access_token = personal_access_token_q.first()
 
             if not personal_access_token:
                 raise codechecker_api_shared.ttypes.RequestFailed(
@@ -794,6 +795,7 @@ def removePersonalAccessToken(self, name):
             session.query(Session) \
                 .filter(Session.id == personal_access_token.auth_session_id) \
                 .delete(synchronize_session=False)
+            personal_access_token_q.delete()
             session.commit()
 
             LOG.info(

web/server/codechecker_server/database/config_db_model.py

@@ -155,14 +155,16 @@ class PersonalAccessToken(Base):
     token = Column(CHAR(32), nullable=False, unique=True)
 
     description = Column(String)
+    # List of group names separated by semicolons.
+    groups = Column(String)
 
     last_access = Column(DateTime, nullable=False)
     expiration = Column(DateTime)
 
     auth_session_id = Column(
         Integer,
-        ForeignKey('auth_sessions.id', deferrable=False, ondelete='CASCADE'),
-        nullable=False)
+        ForeignKey('auth_sessions.id', deferrable=False, ondelete='SET NULL'),
+        nullable=True)
 
     __table_args__ = (
         UniqueConstraint('user_name', 'token_name'),
@@ -174,12 +176,14 @@ def __init__(
         token_name,
         token,
         description,
+        groups,
         auth_session_id
     ):
         self.user_name = user_name
         self.token_name = token_name
         self.token = token
         self.description = description
+        self.groups = groups
         self.auth_session_id = auth_session_id
         self.last_access = datetime.now()
         self.expiration = self.last_access + timedelta(days=365)

web/server/codechecker_server/migrations/config/versions/7ed50f8b3fb8_new_table_for_personal_access_tokens.py

@@ -13,7 +13,6 @@
 import sqlalchemy as sa
 
 
-
 # Revision identifiers, used by Alembic.
 revision = '7ed50f8b3fb8'
 down_revision = '04cd4df82fb5'
@@ -24,19 +23,32 @@
 def upgrade():
     LOG = getLogger("migration/config")
     # ### commands auto generated by Alembic - please adjust! ###
-    op.create_table('personal_access_tokens',
-    sa.Column('id', sa.Integer(), autoincrement=True, nullable=False),
-    sa.Column('user_name', sa.String(), nullable=True),
-    sa.Column('token_name', sa.String(), nullable=True),
-    sa.Column('token', sa.CHAR(length=32), nullable=False),
-    sa.Column('description', sa.String(), nullable=True),
-    sa.Column('last_access', sa.DateTime(), nullable=False),
-    sa.Column('expiration', sa.DateTime(), nullable=True),
-    sa.Column('auth_session_id', sa.Integer(), nullable=False),
-    sa.ForeignKeyConstraint(['auth_session_id'], ['auth_sessions.id'], name=op.f('fk_personal_access_tokens_auth_session_id_auth_sessions'), ondelete='CASCADE'),
-    sa.PrimaryKeyConstraint('id', name=op.f('pk_personal_access_tokens')),
-    sa.UniqueConstraint('token', name=op.f('uq_personal_access_tokens_token')),
-    sa.UniqueConstraint('user_name', 'token_name', name=op.f('uq_personal_access_tokens_user_name'))
+    op.create_table(
+        'personal_access_tokens',
+        sa.Column('id', sa.Integer(), autoincrement=True, nullable=False),
+        sa.Column('user_name', sa.String(), nullable=True),
+        sa.Column('token_name', sa.String(), nullable=True),
+        sa.Column('token', sa.CHAR(length=32), nullable=False),
+        sa.Column('description', sa.String(), nullable=True),
+        sa.Column('groups', sa.String(), nullable=True),
+        sa.Column('last_access', sa.DateTime(), nullable=False),
+        sa.Column('expiration', sa.DateTime(), nullable=True),
+        sa.Column('auth_session_id', sa.Integer(), nullable=True),
+        sa.ForeignKeyConstraint(
+            ['auth_session_id'],
+            ['auth_sessions.id'],
+            name=op.f(
+                'fk_personal_access_tokens_auth_session_id_auth_sessions'),
+            ondelete='SET NULL'),
+        sa.PrimaryKeyConstraint(
+            'id',
+            name=op.f('pk_personal_access_tokens')),
+        sa.UniqueConstraint(
+            'token',
+            name=op.f('uq_personal_access_tokens_token')),
+        sa.UniqueConstraint(
+            'user_name',
+            'token_name', name=op.f('uq_personal_access_tokens_user_name'))
     )
 
     one_year_later = datetime.now() + timedelta(days=365)
@@ -50,8 +62,10 @@ def upgrade():
 
     op.execute(
         f"""
-        INSERT INTO personal_access_tokens (user_name, token_name, token, description, last_access, expiration, auth_session_id)
-        SELECT user_name, {random_string}, token, description, last_access, '{one_year_later}', id
+        INSERT INTO personal_access_tokens (user_name, token_name, token,
+            description, last_access, expiration, auth_session_id)
+        SELECT user_name, {random_string}, token, description, last_access,
+            '{one_year_later}', id
         FROM auth_sessions
         WHERE can_expire = false
         """)

web/server/codechecker_server/session_manager.py

@@ -421,7 +421,8 @@ def __handle_validation(self, auth_string):
         """
         validation = self.__try_auth_dictionary(auth_string) \
             or self.__try_auth_pam(auth_string) \
-            or self.__try_auth_ldap(auth_string)
+            or self.__try_auth_ldap(auth_string) \
+            or self.__try_personal_access_token(auth_string)
         if not validation:
             return False
 
@@ -471,6 +472,34 @@ def __try_auth_token(self, auth_string):
 
         return None
 
+    def __try_personal_access_token(self, auth_string):
+        if not self.__database_connection:
+            return None
+
+        user_name, token = auth_string.split(':', 1)
+
+        transaction = None
+        try:
+            transaction = self.__database_connection()
+            personal_access_token = transaction.query(PersonalAccessToken) \
+                .filter(PersonalAccessToken.user_name == user_name) \
+                .filter(PersonalAccessToken.token == token) \
+                .limit(1).one_or_none()
+        except Exception as e:
+            LOG.error("Couldn't check login in the database:")
+            LOG.error(str(e))
+        finally:
+            if transaction:
+                transaction.close()
+
+        if not personal_access_token:
+            return False
+
+        return {
+            'username': personal_access_token.user_name,
+            'groups': personal_access_token.groups
+        }
+
     def __try_auth_dictionary(self, auth_string):
         """
         Try to authenticate the user against the hardcoded credential list.

web/tests/functional/authentication/test_authentication.py

@@ -98,19 +98,22 @@ def test_privileged_access(self):
         self.assertEqual(user, "cc")
 
         # No personal token in the database.
-        personal_tokens = authd_auth_client.getTokens()
+        personal_tokens = authd_auth_client.getPersonalAccessTokens()
         self.assertEqual(len(personal_tokens), 0)
 
         # Create a new personal token.
         description = "description"
-        personal_token = authd_auth_client.newToken(description)
+        name = "name"
+        personal_token = authd_auth_client.newPersonalAccessToken(
+            name, description)
         token = personal_token.token
         self.assertEqual(personal_token.description, description)
 
         # Check whether the new token has been added.
-        personal_tokens = authd_auth_client.getTokens()
+        personal_tokens = authd_auth_client.getPersonalAccessTokens()
         self.assertEqual(len(personal_tokens), 1)
-        self.assertEqual(personal_tokens[0].token, token)
+        self.assertEqual(personal_tokens[0].token, "")
+        self.assertEqual(personal_tokens[0].name, name)
         self.assertEqual(personal_tokens[0].description, description)
 
         auth_client = env.setup_auth_client(self._test_workspace,
@@ -143,6 +146,9 @@ def test_privileged_access(self):
         self.assertIsNotNone(self.session_token,
                              "Valid credentials didn't give us a token!")
 
+        auth_token_client = \
+            env.setup_auth_client(self._test_workspace,
+                                  session_token=self.session_token)
         user = auth_token_client.getLoggedInUser()
         self.assertEqual(user, "cc")
 
@@ -161,11 +167,11 @@ def test_privileged_access(self):
         auth_client = env.setup_auth_client(self._test_workspace,
                                             session_token=self.session_token)
         # Remove the generated personal token.
-        ret = auth_client.removeToken(token)
+        ret = auth_client.removePersonalAccessToken(name)
         self.assertTrue(ret)
 
         # Check whether no more personal token in the database.
-        personal_tokens = auth_client.getTokens()
+        personal_tokens = auth_client.getPersonalAccessTokens()
         self.assertEqual(len(personal_tokens), 0)
 
         result = auth_client.destroySession()