fix(channelstream): fixed history writing

rhodecode/lib/channelstream.py

@@ -1,370 +1,370 b''
 # Copyright (C) 2016-2023 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program 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 Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 import os
 import itsdangerous
 import logging
 import requests
 import datetime
 
 from dogpile.util.readwrite_lock import ReadWriteMutex
 
 import rhodecode.lib.helpers as h
 from rhodecode.lib.auth import HasRepoPermissionAny
 from rhodecode.lib.ext_json import json
 from rhodecode.model.db import User
 from rhodecode.lib.str_utils import ascii_str
 from rhodecode.lib.hash_utils import sha1_safe
 
 log = logging.getLogger(__name__)
 
 LOCK = ReadWriteMutex()
 
 USER_STATE_PUBLIC_KEYS = [
     'id', 'username', 'first_name', 'last_name',
     'icon_link', 'display_name', 'display_link']
 
 
 class ChannelstreamException(Exception):
     pass
 
 
 class ChannelstreamConnectionException(ChannelstreamException):
     pass
 
 
 class ChannelstreamPermissionException(ChannelstreamException):
     pass
 
 
 def get_channelstream_server_url(config, endpoint):
     return 'http://{}{}'.format(config['server'], endpoint)
 
 
 def channelstream_request(config, payload, endpoint, raise_exc=True):
     signer = itsdangerous.TimestampSigner(config['secret'])
     sig_for_server = signer.sign(endpoint)
     secret_headers = {'x-channelstream-secret': sig_for_server,
                       'x-channelstream-endpoint': endpoint,
                       'Content-Type': 'application/json'}
     req_url = get_channelstream_server_url(config, endpoint)
 
     log.debug('Sending a channelstream request to endpoint: `%s`', req_url)
     response = None
     try:
         response = requests.post(req_url, data=json.dumps(payload),
                                  headers=secret_headers).json()
     except requests.ConnectionError:
         log.exception('ConnectionError occurred for endpoint %s', req_url)
         if raise_exc:
             raise ChannelstreamConnectionException(req_url)
     except Exception:
         log.exception('Exception related to Channelstream happened')
         if raise_exc:
             raise ChannelstreamConnectionException()
     log.debug('Got channelstream response: %s', response)
     return response
 
 
 def get_user_data(user_id):
     user = User.get(user_id)
     return {
         'id': user.user_id,
         'username': user.username,
         'first_name': user.first_name,
         'last_name': user.last_name,
         'icon_link': h.gravatar_url(user.email, 60),
         'display_name': h.person(user, 'username_or_name_or_email'),
         'display_link': h.link_to_user(user),
         'notifications': user.user_data.get('notification_status', True)
     }
 
 
 def broadcast_validator(channel_name):
     """ checks if user can access the broadcast channel """
     if channel_name == 'broadcast':
         return True
 
 
 def repo_validator(channel_name):
     """ checks if user can access the broadcast channel """
     channel_prefix = '/repo$'
     if channel_name.startswith(channel_prefix):
         elements = channel_name[len(channel_prefix):].split('$')
         repo_name = elements[0]
         can_access = HasRepoPermissionAny(
             'repository.read',
             'repository.write',
             'repository.admin')(repo_name)
         log.debug(
             'permission check for %s channel resulted in %s',
             repo_name, can_access)
         if can_access:
             return True
     return False
 
 
 def check_channel_permissions(channels, plugin_validators, should_raise=True):
     valid_channels = []
 
     validators = [broadcast_validator, repo_validator]
     if plugin_validators:
         validators.extend(plugin_validators)
     for channel_name in channels:
         is_valid = False
         for validator in validators:
             if validator(channel_name):
                 is_valid = True
                 break
         if is_valid:
             valid_channels.append(channel_name)
         else:
             if should_raise:
                 raise ChannelstreamPermissionException()
     return valid_channels
 
 
 def get_channels_info(self, channels):
     payload = {'channels': channels}
     # gather persistence info
     return channelstream_request(self._config(), payload, '/info')
 
 
 def parse_channels_info(info_result, include_channel_info=None):
     """
     Returns data that contains only secure information that can be
     presented to clients
     """
     include_channel_info = include_channel_info or []
 
     user_state_dict = {}
     for userinfo in info_result['users']:
         user_state_dict[userinfo['user']] = {
             k: v for k, v in list(userinfo['state'].items())
             if k in USER_STATE_PUBLIC_KEYS
             }
 
     channels_info = {}
 
     for c_name, c_info in list(info_result['channels'].items()):
         if c_name not in include_channel_info:
             continue
         connected_list = []
         for username in c_info['users']:
             connected_list.append({
                 'user': username,
                 'state': user_state_dict[username]
             })
         channels_info[c_name] = {'users': connected_list,
                                  'history': c_info['history']}
 
     return channels_info
 
 
 def log_filepath(history_location, channel_name):
 
     channel_hash = sha1_safe(channel_name, return_type='str')
     filename = f'{channel_hash}.log'
     filepath = os.path.join(history_location, filename)
     return filepath
 
 
 def read_history(history_location, channel_name):
     filepath = log_filepath(history_location, channel_name)
     if not os.path.exists(filepath):
         return []
     history_lines_limit = -100
     history = []
     with open(filepath, 'rb') as f:
         for line in f.readlines()[history_lines_limit:]:
             try:
                 history.append(json.loads(line))
             except Exception:
                 log.exception('Failed to load history')
     return history
 
 
 def update_history_from_logs(config, channels, payload):
     history_location = config.get('history.location')
     for channel in channels:
         history = read_history(history_location, channel)
         payload['channels_info'][channel]['history'] = history
 
 
 def write_history(config, message):
     """ writes a message to a base64encoded filename """
     history_location = config.get('history.location')
     if not os.path.exists(history_location):
         return
     try:
         LOCK.acquire_write_lock()
         filepath = log_filepath(history_location, message['channel'])
         json_message = json.dumps(message)
         with open(filepath, 'ab') as f:
             f.write(json_message)
-            f.write('\n')
+            f.write(b'\n')
     finally:
         LOCK.release_write_lock()
 
 
 def get_connection_validators(registry):
     validators = []
     for k, config in list(registry.rhodecode_plugins.items()):
         validator = config.get('channelstream', {}).get('connect_validator')
         if validator:
             validators.append(validator)
     return validators
 
 
 def get_channelstream_config(registry=None):
     if not registry:
         from pyramid.threadlocal import get_current_registry
         registry = get_current_registry()
 
     rhodecode_plugins = getattr(registry, 'rhodecode_plugins', {})
     channelstream_config = rhodecode_plugins.get('channelstream', {})
     return channelstream_config
 
 
 def post_message(channel, message, username, registry=None):
     channelstream_config = get_channelstream_config(registry)
     if not channelstream_config.get('enabled'):
         return
 
     message_obj = message
     if isinstance(message, str):
         message_obj = {
             'message': message,
             'level': 'success',
             'topic': '/notifications'
         }
 
     log.debug('Channelstream: sending notification to channel %s', channel)
     payload = {
         'type': 'message',
         'timestamp': datetime.datetime.utcnow(),
         'user': 'system',
         'exclude_users': [username],
         'channel': channel,
         'message': message_obj
     }
 
     try:
         return channelstream_request(
             channelstream_config, [payload], '/message',
             raise_exc=False)
     except ChannelstreamException:
         log.exception('Failed to send channelstream data')
         raise
 
 
 def _reload_link(label):
     return (
         '<a onclick="window.location.reload()">'
         '<strong>{}</strong>'
         '</a>'.format(label)
     )
 
 
 def pr_channel(pull_request):
     repo_name = pull_request.target_repo.repo_name
     pull_request_id = pull_request.pull_request_id
     channel = f'/repo${repo_name}$/pr/{pull_request_id}'
     log.debug('Getting pull-request channelstream broadcast channel: %s', channel)
     return channel
 
 
 def comment_channel(repo_name, commit_obj=None, pull_request_obj=None):
     channel = None
     if commit_obj:
         channel = '/repo${}$/commit/{}'.format(
             repo_name, commit_obj.raw_id
         )
     elif pull_request_obj:
         channel = '/repo${}$/pr/{}'.format(
             repo_name, pull_request_obj.pull_request_id
         )
     log.debug('Getting comment channelstream broadcast channel: %s', channel)
 
     return channel
 
 
 def pr_update_channelstream_push(request, pr_broadcast_channel, user, msg, **kwargs):
     """
     Channel push on pull request update
     """
     if not pr_broadcast_channel:
         return
 
     _ = request.translate
 
     message = '{} {}'.format(
         msg,
         _reload_link(_(' Reload page to load changes')))
 
     message_obj = {
         'message': message,
         'level': 'success',
         'topic': '/notifications'
     }
 
     post_message(
         pr_broadcast_channel, message_obj, user.username,
         registry=request.registry)
 
 
 def comment_channelstream_push(request, comment_broadcast_channel, user, msg, **kwargs):
     """
     Channelstream push on comment action, on commit, or pull-request
     """
     if not comment_broadcast_channel:
         return
 
     _ = request.translate
 
     comment_data = kwargs.pop('comment_data', {})
     user_data = kwargs.pop('user_data', {})
     comment_id = list(comment_data.keys())[0] if comment_data else ''
 
     message = '<strong>{}</strong> {} #{}'.format(
         user.username,
         msg,
         comment_id,
     )
 
     message_obj = {
         'message': message,
         'level': 'success',
         'topic': '/notifications'
     }
 
     post_message(
         comment_broadcast_channel, message_obj, user.username,
         registry=request.registry)
 
     message_obj = {
         'message': None,
         'user': user.username,
         'comment_id': comment_id,
         'comment_data': comment_data,
         'user_data': user_data,
         'topic': '/comment'
     }
     post_message(
         comment_broadcast_channel, message_obj, user.username,
         registry=request.registry)