Make repository creation slightly safer

    return os.path.join(repo_path, ".hg", "hgrc")

def create_hgrc_file(app, repository):

def create_hgrc_file(app, repository, repo_path):

    # Since we support both http and https, we set `push_ssl` to False to

    # override the default (which is True) in the Mercurial API.

    # The hg purge extension purges all files and directories not being tracked

    # by Mercurial in the current repository. It will remove unknown files and

    # empty directories. This is not currently used because it is not supported

    # in the Mercurial API.

    repo_path = repository.repo_path(app)

    hgrc_path = get_hgrc_path(repo_path)

    with open(hgrc_path, "w") as fp:

        fp.write("[paths]\n")

import threading

from datetime import date

from galaxy.util import unicodify

log = logging.getLogger(__name__)

new_hgweb_config_template = """

            self.in_memory_config.set("paths", lhs, rhs)

            # Persist our in-memory configuration.

            self.write_config()

        except Exception as e:

            log.debug("Exception in HgWebConfigManager.add_entry(): %s", unicodify(e))

        except Exception:

            log.exception("Exception in HgWebConfigManager.add_entry()")

        finally:

            self.lock.release()

            self.in_memory_config.set("paths", new_lhs, new_rhs)

            # Persist our in-memory configuration.

            self.write_config()

        except Exception as e:

            log.debug("Exception in HgWebConfigManager.change_entry(): %s", unicodify(e))

        except Exception:

            log.exception("Exception in HgWebConfigManager.change_entry()")

        finally:

            self.lock.release()

import logging

import os

import re

import tempfile

from typing import (

    Optional,

    Tuple,

def create_repository_admin_role(app: "ToolShedApp", repository: "Repository"):

    """

    Create a new role with name-spaced name based on the repository name and its owner's public user

    name.  This will ensure that the tole name is unique.

    name.  This will ensure that the role name is unique.

    """

    sa_session = app.model.session

    name = get_repository_admin_role_name(str(repository.name), str(repository.user.username))

    description = "A user or group member with this role can administer this repository."

    role = app.model.Role(name=name, description=description, type=app.model.Role.types.SYSTEM)

    sa_session.add(role)

    session = sa_session()

    with transaction(session):

        session.commit()

    # Associate the role with the repository owner.

    app.model.UserRoleAssociation(repository.user, role)

    # Associate the role with the repository.

    rra = app.model.RepositoryRoleAssociation(repository, role)

    sa_session.add(rra)

    with transaction(session):

        session.commit()

    return role

    type: str,

    description,

    long_description,

    user_id,

    user,

    category_ids=None,

    remote_repository_url=None,

    homepage_url=None,

        homepage_url=homepage_url,

        description=description,

        long_description=long_description,

        user_id=user_id,

        user=user,

    )

    # Flush to get the id.

    sa_session.add(repository)

    session = sa_session()

    with transaction(session):

        session.commit()

    # Create an admin role for the repository.

    create_repository_admin_role(app, repository)

    # Determine the repository's repo_path on disk.

    dir = os.path.join(app.config.file_path, *util.directory_hash_id(repository.id))

    # Create directory if it does not exist.

    if not os.path.exists(dir):

        os.makedirs(dir)

    # Define repo name inside hashed directory.

    repository_path = os.path.join(dir, "repo_%d" % repository.id)

    # Create local repository directory.

    if not os.path.exists(repository_path):

        os.makedirs(repository_path)

    # Create the local repository.

    init_repository(repo_path=repository_path)

    # Add an entry in the hgweb.config file for the local repository.

    lhs = f"{app.config.hgweb_repo_prefix}{repository.user.username}/{repository.name}"

    app.hgweb_config_manager.add_entry(lhs, repository_path)

    # Create a .hg/hgrc file for the local repository.

    create_hgrc_file(app, repository)

    flush_needed = False

    if category_ids:

        # Create category associations

        for category_id in category_ids:

            category = sa_session.query(app.model.Category).get(app.security.decode_id(category_id))

            rca = app.model.RepositoryCategoryAssociation(repository, category)

            sa_session.add(rca)

            flush_needed = True

    if flush_needed:

    # Create an admin role for the repository.

    create_repository_admin_role(app, repository)

    # Create a temporary repo_path on disk.

    with tempfile.TemporaryDirectory(

        dir=app.config.file_path, prefix="f{repository.user.username}-{repository.name}"

    ) as repository_path:

        # Create the local repository.

        init_repository(repo_path=repository_path)

        # Create a .hg/hgrc file for the local repository.

        create_hgrc_file(app, repository, repo_path=repository_path)

        # Add an entry in the hgweb.config file for the local repository.

        lhs = f"{app.config.hgweb_repo_prefix}{repository.user.username}/{repository.name}"

        # Flush to get the id.

        session = sa_session()

        with transaction(session):

            session.commit()

        dir = os.path.join(app.config.file_path, *util.directory_hash_id(repository.id))

        # Define repo name inside hashed directory.

        final_repository_path = os.path.join(dir, "repo_%d" % repository.id)

        # Create final repository directory.

        if not os.path.exists(final_repository_path):

            os.makedirs(final_repository_path)

        os.rename(repository_path, final_repository_path)

    app.hgweb_config_manager.add_entry(lhs, final_repository_path)

    # Update the repository registry.

    app.repository_registry.add_entry(repository)

    message = f"Repository <b>{escape(str(repository.name))}</b> has been created."

            type=repo_type,

            description=synopsis,

            long_description=description,

            user_id=trans.user.id,

            user=trans.user,

            category_ids=category_ids,

            remote_repository_url=remote_repository_url,

            homepage_url=homepage_url,

                    repository_type,

                    description,

                    long_description,

                    user_id=trans.user.id,

                    user=trans.user,

                    category_ids=category_ids,

                    remote_repository_url=remote_repository_url,

                    homepage_url=homepage_url,