# Copyright (C) 2010-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/ """ Database creation, and setup module for RhodeCode Enterprise. Used for creation of database as well as for migration operations """ import os import sys import time import uuid import logging import getpass from os.path import dirname as dn, join as jn from sqlalchemy.engine import create_engine from rhodecode import __dbversion__ from rhodecode.model import init_model from rhodecode.model.user import UserModel from rhodecode.model.db import ( User, Permission, RhodeCodeUi, RhodeCodeSetting, UserToPerm, DbMigrateVersion, RepoGroup, UserRepoGroupToPerm, CacheKey, Repository) from rhodecode.model.meta import Session, Base from rhodecode.model.permission import PermissionModel from rhodecode.model.repo import RepoModel from rhodecode.model.repo_group import RepoGroupModel from rhodecode.model.settings import SettingsModel log = logging.getLogger(__name__) def notify(msg): """ Notification for migrations messages """ ml = len(msg) + (4 * 2) print((('\n%s\n*** %s ***\n%s' % ('*' * ml, msg, '*' * ml)).upper())) class DbManage(object): def __init__(self, log_sql, dbconf, root, tests=False, SESSION=None, cli_args=None, enc_key=b''): self.dbname = dbconf.split('/')[-1] self.tests = tests self.root = root self.dburi = dbconf self.log_sql = log_sql self.cli_args = cli_args or {} self.sa = None self.engine = None self.enc_key = enc_key # sets .sa .engine self.init_db(SESSION=SESSION) self.ask_ok = self.get_ask_ok_func(self.cli_args.get('force_ask')) def db_exists(self): if not self.sa: self.init_db() try: self.sa.query(RhodeCodeUi)\ .filter(RhodeCodeUi.ui_key == '/')\ .scalar() return True except Exception: return False finally: self.sa.rollback() def get_ask_ok_func(self, param): if param not in [None]: # return a function lambda that has a default set to param return lambda *args, **kwargs: param else: from rhodecode.lib.utils import ask_ok return ask_ok def init_db(self, SESSION=None): if SESSION: self.sa = SESSION self.engine = SESSION.bind else: # init new sessions engine = create_engine(self.dburi, echo=self.log_sql) init_model(engine, encryption_key=self.enc_key) self.sa = Session() self.engine = engine def create_tables(self, override=False): """ Create a auth database """ log.info("Existing database with the same name is going to be destroyed.") log.info("Setup command will run DROP ALL command on that database.") engine = self.engine if self.tests: destroy = True else: destroy = self.ask_ok('Are you sure that you want to destroy the old database? [y/n]') if not destroy: log.info('db tables bootstrap: Nothing done.') sys.exit(0) if destroy: Base.metadata.drop_all(bind=engine) checkfirst = not override Base.metadata.create_all(bind=engine, checkfirst=checkfirst) log.info('Created tables for %s', self.dbname) def set_db_version(self): ver = DbMigrateVersion() ver.version = __dbversion__ ver.repository_id = 'rhodecode_db_migrations' ver.repository_path = 'versions' self.sa.add(ver) log.info('db version set to: %s', __dbversion__) def run_post_migration_tasks(self): """ Run various tasks before actually doing migrations """ # delete cache keys on each upgrade total = CacheKey.query().count() log.info("Deleting (%s) cache keys now...", total) CacheKey.delete_all_cache() def upgrade(self, version=None): """ Upgrades given database schema to given revision following all needed steps, to perform the upgrade """ from rhodecode.lib.dbmigrate.migrate.versioning import api from rhodecode.lib.dbmigrate.migrate.exceptions import DatabaseNotControlledError if 'sqlite' in self.dburi: print( '********************** WARNING **********************\n' 'Make sure your version of sqlite is at least 3.7.X. \n' 'Earlier versions are known to fail on some migrations\n' '*****************************************************\n') upgrade = self.ask_ok( 'You are about to perform a database upgrade. Make ' 'sure you have backed up your database. ' 'Continue ? [y/n]') if not upgrade: log.info('No upgrade performed') sys.exit(0) repository_path = jn(dn(dn(dn(os.path.realpath(__file__)))), 'rhodecode/lib/dbmigrate') db_uri = self.dburi if version: DbMigrateVersion.set_version(version) try: curr_version = api.db_version(db_uri, repository_path) msg = (f'Found current database db_uri under version ' f'control with version {curr_version}') except (RuntimeError, DatabaseNotControlledError): curr_version = 1 msg = f'Current database is not under version control. ' \ f'Setting as version {curr_version}' api.version_control(db_uri, repository_path, curr_version) notify(msg) if curr_version == __dbversion__: log.info('This database is already at the newest version') sys.exit(0) upgrade_steps = list(range(curr_version + 1, __dbversion__ + 1)) notify(f'attempting to upgrade database from ' f'version {curr_version} to version {__dbversion__}') # CALL THE PROPER ORDER OF STEPS TO PERFORM FULL UPGRADE _step = None for step in upgrade_steps: notify(f'performing upgrade step {step}') time.sleep(0.5) api.upgrade(db_uri, repository_path, step) self.sa.rollback() notify(f'schema upgrade for step {step} completed') _step = step self.run_post_migration_tasks() notify(f'upgrade to version {step} successful') def fix_repo_paths(self): """ Fixes an old RhodeCode version path into new one without a '*' """ paths = self.sa.query(RhodeCodeUi)\ .filter(RhodeCodeUi.ui_key == '/')\ .scalar() paths.ui_value = paths.ui_value.replace('*', '') try: self.sa.add(paths) self.sa.commit() except Exception: self.sa.rollback() raise def fix_default_user(self): """ Fixes an old default user with some 'nicer' default values, used mostly for anonymous access """ def_user = self.sa.query(User)\ .filter(User.username == User.DEFAULT_USER)\ .one() def_user.name = 'Anonymous' def_user.lastname = 'User' def_user.email = User.DEFAULT_USER_EMAIL try: self.sa.add(def_user) self.sa.commit() except Exception: self.sa.rollback() raise def fix_settings(self): """ Fixes rhodecode settings and adds ga_code key for google analytics """ hgsettings3 = RhodeCodeSetting('ga_code', '') try: self.sa.add(hgsettings3) self.sa.commit() except Exception: self.sa.rollback() raise def create_admin_and_prompt(self): # defaults defaults = self.cli_args username = defaults.get('username') password = defaults.get('password') email = defaults.get('email') if username is None: username = input('Specify admin username:') if password is None: password = self._get_admin_password() if not password: # second try password = self._get_admin_password() if not password: sys.exit() if email is None: email = input('Specify admin email:') api_key = self.cli_args.get('api_key') self.create_user(username, password, email, True, strict_creation_check=False, api_key=api_key) def _get_admin_password(self): password = getpass.getpass('Specify admin password ' '(min 6 chars):') confirm = getpass.getpass('Confirm password:') if password != confirm: log.error('passwords mismatch') return False if len(password) < 6: log.error('password is too short - use at least 6 characters') return False return password def create_test_admin_and_users(self): log.info('creating admin and regular test users') from rhodecode.tests import TEST_USER_ADMIN_LOGIN, \ TEST_USER_ADMIN_PASS, TEST_USER_ADMIN_EMAIL, \ TEST_USER_REGULAR_LOGIN, TEST_USER_REGULAR_PASS, \ TEST_USER_REGULAR_EMAIL, TEST_USER_REGULAR2_LOGIN, \ TEST_USER_REGULAR2_PASS, TEST_USER_REGULAR2_EMAIL self.create_user(TEST_USER_ADMIN_LOGIN, TEST_USER_ADMIN_PASS, TEST_USER_ADMIN_EMAIL, True, api_key=True) self.create_user(TEST_USER_REGULAR_LOGIN, TEST_USER_REGULAR_PASS, TEST_USER_REGULAR_EMAIL, False, api_key=True) self.create_user(TEST_USER_REGULAR2_LOGIN, TEST_USER_REGULAR2_PASS, TEST_USER_REGULAR2_EMAIL, False, api_key=True) def create_ui_settings(self, repo_store_path): """ Creates ui settings, fills out hooks and disables dotencode """ settings_model = SettingsModel(sa=self.sa) from rhodecode.lib.vcs.backends.hg import largefiles_store from rhodecode.lib.vcs.backends.git import lfs_store # Build HOOKS hooks = [ (RhodeCodeUi.HOOK_REPO_SIZE, 'python:vcsserver.hooks.repo_size'), # HG (RhodeCodeUi.HOOK_PRE_PULL, 'python:vcsserver.hooks.pre_pull'), (RhodeCodeUi.HOOK_PULL, 'python:vcsserver.hooks.log_pull_action'), (RhodeCodeUi.HOOK_PRE_PUSH, 'python:vcsserver.hooks.pre_push'), (RhodeCodeUi.HOOK_PRETX_PUSH, 'python:vcsserver.hooks.pre_push'), (RhodeCodeUi.HOOK_PUSH, 'python:vcsserver.hooks.log_push_action'), (RhodeCodeUi.HOOK_PUSH_KEY, 'python:vcsserver.hooks.key_push'), ] for key, value in hooks: hook_obj = settings_model.get_ui_by_key(key) hooks2 = hook_obj if hook_obj else RhodeCodeUi() hooks2.ui_section = 'hooks' hooks2.ui_key = key hooks2.ui_value = value self.sa.add(hooks2) # enable largefiles largefiles = RhodeCodeUi() largefiles.ui_section = 'extensions' largefiles.ui_key = 'largefiles' largefiles.ui_value = '' self.sa.add(largefiles) # set default largefiles cache dir, defaults to # /repo_store_location/.cache/largefiles largefiles = RhodeCodeUi() largefiles.ui_section = 'largefiles' largefiles.ui_key = 'usercache' largefiles.ui_value = largefiles_store(repo_store_path) self.sa.add(largefiles) # set default lfs cache dir, defaults to # /repo_store_location/.cache/lfs_store lfsstore = RhodeCodeUi() lfsstore.ui_section = 'vcs_git_lfs' lfsstore.ui_key = 'store_location' lfsstore.ui_value = lfs_store(repo_store_path) self.sa.add(lfsstore) # enable hgevolve disabled by default hgevolve = RhodeCodeUi() hgevolve.ui_section = 'extensions' hgevolve.ui_key = 'evolve' hgevolve.ui_value = '' hgevolve.ui_active = False self.sa.add(hgevolve) hgevolve = RhodeCodeUi() hgevolve.ui_section = 'experimental' hgevolve.ui_key = 'evolution' hgevolve.ui_value = '' hgevolve.ui_active = False self.sa.add(hgevolve) hgevolve = RhodeCodeUi() hgevolve.ui_section = 'experimental' hgevolve.ui_key = 'evolution.exchange' hgevolve.ui_value = '' hgevolve.ui_active = False self.sa.add(hgevolve) hgevolve = RhodeCodeUi() hgevolve.ui_section = 'extensions' hgevolve.ui_key = 'topic' hgevolve.ui_value = '' hgevolve.ui_active = False self.sa.add(hgevolve) # enable hggit disabled by default hggit = RhodeCodeUi() hggit.ui_section = 'extensions' hggit.ui_key = 'hggit' hggit.ui_value = '' hggit.ui_active = False self.sa.add(hggit) # set svn branch defaults branches = ["/branches/*", "/trunk"] tags = ["/tags/*"] for branch in branches: settings_model.create_ui_section_value( RhodeCodeUi.SVN_BRANCH_ID, branch) for tag in tags: settings_model.create_ui_section_value(RhodeCodeUi.SVN_TAG_ID, tag) def create_auth_plugin_options(self, skip_existing=False): """ Create default auth plugin settings, and make it active :param skip_existing: """ defaults = [ ('auth_plugins', 'egg:rhodecode-enterprise-ce#token,egg:rhodecode-enterprise-ce#rhodecode', 'list'), ('auth_authtoken_enabled', 'True', 'bool'), ('auth_rhodecode_enabled', 'True', 'bool'), ] for k, v, t in defaults: if (skip_existing and SettingsModel().get_setting_by_name(k) is not None): log.debug('Skipping option %s', k) continue setting = RhodeCodeSetting(k, v, t) self.sa.add(setting) def create_default_options(self, skip_existing=False): """Creates default settings""" for k, v, t in [ ('default_repo_enable_locking', False, 'bool'), ('default_repo_enable_downloads', False, 'bool'), ('default_repo_enable_statistics', False, 'bool'), ('default_repo_private', False, 'bool'), ('default_repo_type', 'hg', 'unicode')]: if (skip_existing and SettingsModel().get_setting_by_name(k) is not None): log.debug('Skipping option %s', k) continue setting = RhodeCodeSetting(k, v, t) self.sa.add(setting) def fixup_groups(self): def_usr = User.get_default_user() for g in RepoGroup.query().all(): g.group_name = g.get_new_name(g.name) self.sa.add(g) # get default perm default = UserRepoGroupToPerm.query()\ .filter(UserRepoGroupToPerm.group == g)\ .filter(UserRepoGroupToPerm.user == def_usr)\ .scalar() if default is None: log.debug('missing default permission for group %s adding', g) perm_obj = RepoGroupModel()._create_default_perms(g) self.sa.add(perm_obj) def reset_permissions(self, username): """ Resets permissions to default state, useful when old systems had bad permissions, we must clean them up :param username: """ default_user = User.get_by_username(username) if not default_user: return u2p = UserToPerm.query()\ .filter(UserToPerm.user == default_user).all() fixed = False if len(u2p) != len(Permission.DEFAULT_USER_PERMISSIONS): for p in u2p: Session().delete(p) fixed = True self.populate_default_permissions() return fixed def config_prompt(self, test_repo_path='', retries=3): defaults = self.cli_args _path = defaults.get('repos_location') if retries == 3: log.info('Setting up repositories config') if _path is not None: path = _path elif not self.tests and not test_repo_path: path = input( 'Enter a valid absolute path to store repositories. ' 'All repositories in that path will be added automatically:' ) else: path = test_repo_path path_ok = True # check proper dir if not os.path.isdir(path): path_ok = False log.error('Given path %s is not a valid directory', path) elif not os.path.isabs(path): path_ok = False log.error('Given path %s is not an absolute path', path) # check if path is at least readable. if not os.access(path, os.R_OK): path_ok = False log.error('Given path %s is not readable', path) # check write access, warn user about non writeable paths elif not os.access(path, os.W_OK) and path_ok: log.warning('No write permission to given path %s', path) q = (f'Given path {path} is not writeable, do you want to ' f'continue with read only mode ? [y/n]') if not self.ask_ok(q): log.error('Canceled by user') sys.exit(-1) if retries == 0: sys.exit('max retries reached') if not path_ok: retries -= 1 return self.config_prompt(test_repo_path, retries) real_path = os.path.normpath(os.path.realpath(path)) if real_path != os.path.normpath(path): q = (f'Path looks like a symlink, RhodeCode Enterprise will store ' f'given path as {real_path} ? [y/n]') if not self.ask_ok(q): log.error('Canceled by user') sys.exit(-1) return real_path def create_settings(self, path): self.create_ui_settings(path) ui_config = [ ('web', 'push_ssl', 'False'), ('web', 'allow_archive', 'gz zip bz2'), ('web', 'allow_push', '*'), ('web', 'baseurl', '/'), ('paths', '/', path), ('phases', 'publish', 'True') ] for section, key, value in ui_config: ui_conf = RhodeCodeUi() setattr(ui_conf, 'ui_section', section) setattr(ui_conf, 'ui_key', key) setattr(ui_conf, 'ui_value', value) self.sa.add(ui_conf) # rhodecode app settings settings = [ ('realm', 'RhodeCode', 'unicode'), ('title', '', 'unicode'), ('pre_code', '', 'unicode'), ('post_code', '', 'unicode'), # Visual ('show_public_icon', True, 'bool'), ('show_private_icon', True, 'bool'), ('stylify_metatags', True, 'bool'), ('dashboard_items', 100, 'int'), ('admin_grid_items', 25, 'int'), ('markup_renderer', 'markdown', 'unicode'), ('repository_fields', True, 'bool'), ('show_version', True, 'bool'), ('show_revision_number', True, 'bool'), ('show_sha_length', 12, 'int'), ('use_gravatar', False, 'bool'), ('gravatar_url', User.DEFAULT_GRAVATAR_URL, 'unicode'), ('clone_uri_tmpl', Repository.DEFAULT_CLONE_URI, 'unicode'), ('clone_uri_id_tmpl', Repository.DEFAULT_CLONE_URI_ID, 'unicode'), ('clone_uri_ssh_tmpl', Repository.DEFAULT_CLONE_URI_SSH, 'unicode'), ('support_url', '', 'unicode'), ('update_url', RhodeCodeSetting.DEFAULT_UPDATE_URL, 'unicode'), # VCS Settings ('pr_merge_enabled', True, 'bool'), ('use_outdated_comments', True, 'bool'), ('diff_cache', True, 'bool'), ] for key, val, type_ in settings: sett = RhodeCodeSetting(key, val, type_) self.sa.add(sett) self.create_auth_plugin_options() self.create_default_options() log.info('created ui config') def create_user(self, username, password, email='', admin=False, strict_creation_check=True, api_key=None): log.info('creating user `%s`', username) user = UserModel().create_or_update( username, password, email, firstname='RhodeCode', lastname='Admin', active=True, admin=admin, extern_type="rhodecode", strict_creation_check=strict_creation_check) if api_key: log.info('setting a new default auth token for user `%s`', username) UserModel().add_auth_token( user=user, lifetime_minutes=-1, role=UserModel.auth_token_role.ROLE_ALL, description='BUILTIN TOKEN') def create_default_user(self): log.info('creating default user') # create default user for handling default permissions. user = UserModel().create_or_update(username=User.DEFAULT_USER, password=str(uuid.uuid1())[:20], email=User.DEFAULT_USER_EMAIL, firstname='Anonymous', lastname='User', strict_creation_check=False) # based on configuration options activate/de-activate this user which # controls anonymous access if self.cli_args.get('public_access') is False: log.info('Public access disabled') user.active = False Session().add(user) Session().commit() def create_permissions(self): """ Creates all permissions defined in the system """ # module.(access|create|change|delete)_[name] # module.(none|read|write|admin) log.info('creating permissions') PermissionModel(self.sa).create_permissions() def populate_default_permissions(self): """ Populate default permissions. It will create only the default permissions that are missing, and not alter already defined ones """ log.info('creating default user permissions') PermissionModel(self.sa).create_default_user_permissions(user=User.DEFAULT_USER)