Скрипт nexus-migrator.py

 
#!/usr/bin/env python3"""Script to transfer Nexus repositories, users, roles, and permissions between instances"""import requestsimport jsonimport base64import argparseimport sysfrom typing import Dict, List, Anyclass NexusMigrator:    def __init__(self, source_url: str, source_auth: tuple,                 target_url: str, target_auth: tuple, verify_ssl: bool = True):        self.source_url = source_url.rstrip('/')        self.target_url = target_url.rstrip('/')        self.source_auth = source_auth        self.target_auth = target_auth        self.verify_ssl = verify_ssl        if not verify_ssl:            import urllib3            urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)    def _make_request(self, url: str, auth: tuple, method: str = 'GET',                     data: Dict = None, headers: Dict = None) -> Dict:        """Make HTTP request to Nexus API"""        try:            default_headers = {'Content-Type': 'application/json'}            if headers:                default_headers.update(headers)            response = requests.request(                method=method,                url=url,                auth=auth,                json=data,                headers=default_headers,                verify=self.verify_ssl            )            response.raise_for_status()            return response.json() if response.content else {"result": "success"}        except requests.exceptions.RequestException as e:            if response.status_code != 404:                print(f"Error making request to {url}: {e} Response: {response.json()}")            return {}    def get_repositories(self) -> List[Dict]:        """Get all repositories from source Nexus"""        print("Fetching repositories from source Nexus...")        url = f"{self.source_url}/service/rest/v1/repositories"        result = self._make_request(url, self.source_auth)        return result    def get_repository_details(self, repo_config: Dict) -> Dict:        """Get detailed configuration for a specific repository"""        repo_name = repo_config.get('name')        repo_format = 'maven' if repo_config.get('format') == 'maven2' else repo_config.get('format')        repo_type = repo_config.get('type')        url = f"{self.source_url}/service/rest/v1/repositories/{repo_format}/{repo_type}/{repo_name}"        return self._make_request(url, self.source_auth)        def check_repository(self, repo_config: Dict) -> bool:        """Check if repository exists in target Nexus"""        repo_name = repo_config.get('name')        repo_format = 'maven' if repo_config.get('format') == 'maven2' else repo_config.get('format')        repo_type = repo_config.get('type')        url = f"{self.target_url}/service/rest/v1/repositories/{repo_format}/{repo_type}/{repo_name}"        result = self._make_request(url, self.target_auth)        if result:            print(f" Repository {repo_name} already exists, skipping...")            return True        else:            return False    def create_repository(self, repo_config: Dict) -> bool:        """Create repository in target Nexus"""        repo_name = repo_config.get('name')        repo_format = 'maven' if repo_config.get('format') == 'maven2' else repo_config.get('format')        repo_type = repo_config.get('type')        url = f"{self.target_url}/service/rest/v1/repositories/{repo_format}/{repo_type}"        # Clean up config for creation        config = repo_config.copy()        config.pop('format', None)        config.pop('type', None)        if repo_type == 'proxy' and config['httpClient']['authentication'] is not None:            config['httpClient']['authentication']['password'] = '' # You'll need to handle passwords properly                    result = self._make_request(url, self.target_auth, 'POST', config)        if result:            print(f"Created repository: {repo_name}")            return True        else:            print(f"Failed to create repository: {repo_name}")            return False    def get_roles(self) -> List[Dict]:        """Get all roles from source Nexus"""        print("Fetching roles from source Nexus...")        url = f"{self.source_url}/service/rest/v1/security/roles"        result = self._make_request(url, self.source_auth)        return result    def check_role(self, role_config: Dict) -> bool:        role_id = role_config.get('id')        url = f"{self.target_url}/service/rest/v1/security/roles/{role_id}"        result = self._make_request(url, self.target_auth)        if result:            print(f" Role {role_id} already exists, skipping...")            return True        else:            return False    def create_role(self, role_config: Dict) -> bool:        """Create role in target Nexus"""        role_id = role_config.get('id')        url = f"{self.target_url}/service/rest/v1/security/roles"        result = self._make_request(url, self.target_auth, 'POST', role_config)        if result:            print(f"Created role: {role_id}")            return True        else:            print(f"Failed to create role: {role_id}")            return False    def get_users(self) -> List[Dict]:        """Get all users from source Nexus"""        print("Fetching users from source Nexus...")        url = f"{self.source_url}/service/rest/v1/security/users"        result = self._make_request(url, self.source_auth)        return result        def check_user(self, user_config: Dict) -> bool:        user_id = user_config.get('userId')        url = f"{self.target_url}/service/rest/v1/security/users"        users = self._make_request(url, self.target_auth)        for user in users:            if user.get('userId') == user_id:                print(f" User {user_id} already exists, skipping...")                return True        return False    def create_user(self, user_config: Dict) -> bool:        """Create user in target Nexus"""        user_id = user_config.get('userId')        url = f"{self.target_url}/service/rest/v1/security/users"        # Prepare user data for creation        user_data = {            'userId': user_config.get('userId'),            'firstName': user_config.get('firstName', ''),            'lastName': user_config.get('lastName', ''),            'emailAddress': user_config.get('emailAddress', ''),            'password': 'tempPassword123!',  # You'll need to handle passwords properly            'status': user_config.get('status', 'active'),            'roles': user_config.get('roles') if len(user_config.get('roles')) > 0 else ['nx-anonymous']        }        result = self._make_request(url, self.target_auth, 'POST', user_data)        if result:            print(f" Created user: {user_id}")            return True        else:            print(f" Failed to create user: {user_id}")            return False    def update_user(self, user_config: Dict) -> bool:        """Update user in target Nexus"""        user_id = user_config.get('userId')        url = f"{self.target_url}/service/rest/v1/security/users/{user_id}"        # Prepare user data for update        user_data = {            'userId': user_config.get('userId'),            'firstName': user_config.get('firstName', ''),            'lastName': user_config.get('lastName', ''),            'emailAddress': user_config.get('emailAddress', ''),            'password': 'tempPassword123!',  # You'll need to handle passwords properly            'status': user_config.get('status', 'active'),            'source': user_config.get('source', 'default'),            'roles': user_config.get('roles') if len(user_config.get('roles')) > 0 else ['nx-anonymous']        }        result = self._make_request(url, self.target_auth, 'PUT', user_data)        if result:            print(f" Updated user: {user_id}")            return True        else:            print(f" Failed to update user: {user_id}")            return False    def get_privileges(self) -> List[Dict]:        """Get all privileges from source Nexus"""        print("Fetching privileges from source Nexus...")        url = f"{self.source_url}/service/rest/v1/security/privileges"        result = self._make_request(url, self.source_auth)        return result        def check_privilege(self, privilege_config: Dict) -> bool:        privilege_name = privilege_config.get('name')        url = f"{self.target_url}/service/rest/v1/security/privileges/{privilege_name}"        result = self._make_request(url, self.target_auth)        if result:            print(f" Privilege {privilege_name} already exists, skipping...")            return True        else:            return False    def create_privilege(self, privilege_config: Dict) -> bool:        """Create privilege in target Nexus"""        privilege_name = privilege_config.get('name')        privilege_type = privilege_config.get('type')        url = f"{self.target_url}/service/rest/v1/security/privileges/{privilege_type}"        result = self._make_request(url, self.target_auth, 'POST', privilege_config)        if result:            print(f" Created privilege: {privilege_name}")            return True        else:            print(f" Failed to create privilege: {privilege_name}")            return False    def migrate_repositories(self) -> bool:        """Migrate all repositories"""        print("\n=== Migrating Repositories ===")        repositories = self.get_repositories()        skip_count = 0        success_count = 0        for repo in repositories:            repo_name = repo.get('name')            print(f"Processing repository: {repo_name}")            # Get detailed repository configuration            repo_details = self.get_repository_details(repo)            if repo_details and self.check_repository(repo_details):                skip_count += 1                continue            if repo_details and self.create_repository(repo_details):                success_count += 1        print(f"Repository migration -  created: {success_count}, skipped existing: {skip_count} out of {len(repositories)} total")        return success_count > 0    def migrate_roles(self) -> bool:        """Migrate all roles"""        print("\n=== Migrating Roles ===")        roles = self.get_roles()        print("Processing roles")        skip_builtin_count = 0        skip_count = 0        success_count = 0        for role in roles:            role_id = role.get('id')            if role_id and role_id.startswith('nx-'):  # Skip built-in roles                skip_builtin_count += 1                continue            if role_id and self.check_role(role):                skip_count += 1                continue            if self.create_role(role):                success_count += 1        print(f"Role migration - created: {success_count}, skipped built-in: {skip_builtin_count}, skipped existing: {skip_count} out of {len(roles)} total")        return success_count > 0    def migrate_users(self) -> bool:        """Migrate all users"""        print("\n=== Migrating Users ===")        users = self.get_users()        skip_count = 0        update_count = 0        success_count = 0        for user in users:            user_id = user.get('userId')            print(f"Processing user: {user_id}")            if user_id and self.check_user(user):                if user_id == 'dev-platform-anonymous':                    if self.update_user(user):                        update_count += 1                else:                    skip_count += 1                continue            if self.create_user(user):                success_count += 1        print(f"User migration - created: {success_count}, updated existing: {update_count}, skipped existing: {skip_count} out of {len(users)} total")        return success_count > 0    def migrate_privileges(self) -> bool:        """Migrate all privileges"""        print("\n=== Migrating Privileges ===")        privileges = self.get_privileges()                print("Processing privileges")        skip_builtin_count = 0        skip_count = 0        success_count = 0        for privilege in privileges:            privilege_name = privilege.get('name')            if privilege_name and privilege_name.startswith('nx-'):  # Skip built-in privileges                skip_builtin_count += 1                continue            if privilege_name and self.check_privilege(privilege):                skip_count += 1                continue            if self.create_privilege(privilege):                success_count += 1        print(f"Privilege migration -  created: {success_count}, skipped built-in: {skip_builtin_count}, skipped existing: {skip_count} out of {len(privileges)} total")        return success_count > 0    def full_migration(self):        """Perform full migration"""        print("Starting Nexus migration...")        # Order matters: repositories -> privileges -> roles -> users        self.migrate_repositories()        self.migrate_privileges()        self.migrate_roles()        self.migrate_users()        print("\n=== Migration Complete ===")def main():    parser = argparse.ArgumentParser(description='Migrate Nexus repositories, users, roles, and privileges')    parser.add_argument('--source-url', required=True, help='Source Nexus URL')    parser.add_argument('--source-user', required=True, help='Source Nexus username')    parser.add_argument('--source-password', required=True, help='Source Nexus password')    parser.add_argument('--target-url', required=True, help='Target Nexus URL')    parser.add_argument('--target-user', required=True, help='Target Nexus username')    parser.add_argument('--target-password', required=True, help='Target Nexus password')    parser.add_argument('--no-ssl-verify', action='store_true', help='Disable SSL verification')    args = parser.parse_args()    source_auth = (args.source_user, args.source_password)    target_auth = (args.target_user, args.target_password)    migrator = NexusMigrator(        source_url=args.source_url,        source_auth=source_auth,        target_url=args.target_url,        target_auth=target_auth,        verify_ssl=not args.no_ssl_verify    )    migrator.full_migration()if __name__ == "__main__":    main()