codekingpro/portable-devtools
115k
1#########################################################################2#3# pgAdmin 4 - PostgreSQL Tools4#5# Copyright (C) 2013 - 2024, The pgAdmin Development Team6# This software is released under the PostgreSQL Licence7#8##########################################################################9 10"""Perform the initial setup of the application, by creating the auth11and settings database."""12 13import os14import sys15import typer16from rich.console import Console17from rich.table import Table18from rich import box, print19import json as jsonlib20from functools import wraps21 22console = Console()23app = typer.Typer()24 25# We need to include the root directory in sys.path to ensure that we can26# find everything we need when running in the standalone runtime.27root = os.path.dirname(os.path.realpath(__file__))28if sys.path[0] != root:29 sys.path.insert(0, root)30 31import builtins32import config33 34# Grab the SERVER_MODE if it's been set by the runtime35if 'SERVER_MODE' in globals():36 builtins.SERVER_MODE = globals()['SERVER_MODE']37else:38 builtins.SERVER_MODE = None39 40from pgadmin.model import db, Version, User, \41 SCHEMA_VERSION as CURRENT_SCHEMA_VERSION42from pgadmin import create_app43from pgadmin.utils import clear_database_servers, dump_database_servers, \44 load_database_servers, _handle_error45from pgadmin.setup import db_upgrade, create_app_data_directory46from typing import Optional, List47from typing_extensions import Annotated48from pgadmin.utils.constants import INTERNAL, LDAP, OAUTH2, \49 KERBEROS, WEBSERVER50from pgadmin.tools.user_management import create_user, delete_user, \51 update_user as user_management_update_user52from enum import Enum53from flask_babel import gettext54 55 56app = typer.Typer(pretty_exceptions_show_locals=False)57 58 59def update_sqlite_path(f):60 """61 This function will behave as a decorator which will check if62 sqlite path is provided and will update it in config63 """64 65 @wraps(f)66 def wrap(*args, **kwargs):67 if kwargs and kwargs.get('sqlite_path') is not None:68 # update the sqlite path69 config.SQLITE_PATH = kwargs['sqlite_path']70 return f(*args, **kwargs)71 72 return wrap73 74 75class ManageServers:76 77 @app.command()78 @update_sqlite_path79 def dump_servers(output_file: str, user: Optional[str] = None,80 auth_source: Optional[str] = INTERNAL,81 sqlite_path: Optional[str] = None,82 server: List[int] = None):83 """Dump the server groups and servers. """84 85 # What user?86 dump_user = user if user is not None else config.DESKTOP_USER87 88 print('----------')89 print('Dumping servers with:')90 print('User:', dump_user)91 print('SQLite pgAdmin config:', config.SQLITE_PATH)92 print('----------')93 94 try:95 app = create_app(config.APP_NAME + '-cli')96 with app.test_request_context():97 dump_database_servers(output_file, server, dump_user, True,98 auth_source)99 except Exception as e:100 print(str(e))101 102 @app.command()103 @update_sqlite_path104 def load_servers(input_file: str, user: Optional[str] = None,105 auth_source: Optional[str] = INTERNAL,106 sqlite_path: Optional[str] = None,107 replace: Optional[bool] = False108 ):109 110 """Load server groups and servers."""111 112 # What user?113 load_user = user if user is not None else config.DESKTOP_USER114 115 print('----------')116 print('Loading servers with:')117 print('User:', load_user)118 print('SQLite pgAdmin config:', config.SQLITE_PATH)119 print('----------')120 121 try:122 app = create_app(config.APP_NAME + '-cli')123 with app.test_request_context():124 if replace:125 clear_database_servers(load_user, True, auth_source)126 load_database_servers(input_file, None, load_user, True,127 auth_source)128 except Exception as e:129 print(str(e))130 131 132class AuthExtTypes(str, Enum):133 oauth2 = OAUTH2134 ldap = LDAP135 kerberos = KERBEROS136 webserver = WEBSERVER137 138 139# Enum class can not be extended140class AuthType(str, Enum):141 oauth2 = OAUTH2142 ldap = LDAP143 kerberos = KERBEROS144 webserver = WEBSERVER145 internal = INTERNAL146 147 148class ManageUsers:149 150 @app.command()151 @update_sqlite_path152 def add_user(email: str, password: str,153 role: Annotated[Optional[bool], typer.Option(154 "--admin/--nonadmin")] = False,155 active: Annotated[Optional[bool],156 typer.Option("--active/--inactive")] = True,157 console: Optional[bool] = True,158 json: Optional[bool] = False,159 sqlite_path: Optional[str] = None,160 ):161 """Add Internal user. """162 163 data = {164 'email': email,165 'role': 1 if role else 2,166 'active': active,167 'auth_source': INTERNAL,168 'newPassword': password,169 'confirmPassword': password,170 }171 ManageUsers.create_user(data, console, json)172 173 @app.command()174 @update_sqlite_path175 def add_external_user(username: str,176 auth_source: AuthExtTypes = AuthExtTypes.oauth2,177 email: Optional[str] = None,178 role: Annotated[Optional[bool],179 typer.Option(180 "--admin/--nonadmin")] = False,181 active: Annotated[Optional[bool],182 typer.Option(183 "--active/--inactive")] = True,184 console: Optional[bool] = True,185 json: Optional[bool] = False,186 sqlite_path: Optional[str] = None,187 ):188 """Add external user, other than Internal like189 Ldap, Ouath2, Kerberos, Webserver. """190 191 data = {192 'username': username,193 'email': email,194 'role': 1 if role else 2,195 'active': active,196 'auth_source': auth_source197 }198 ManageUsers.create_user(data, console, json)199 200 @app.command()201 @update_sqlite_path202 def delete_user(username: str,203 auth_source: AuthType = AuthType.internal,204 auto_confirm: Annotated[Optional[bool],205 typer.Option(206 "--yes")] = False,207 sqlite_path: Optional[str] = None,208 ):209 """Delete the user. """210 211 confirm_msg = "Are you sure you want to delete it?"212 213 if auto_confirm or typer.confirm(confirm_msg):214 app = create_app(config.APP_NAME + '-cli')215 with app.test_request_context():216 uid = ManageUsers.get_user(username=username,217 auth_source=auth_source)218 if not uid:219 print("User not found")220 else:221 status, msg = delete_user(uid)222 if status:223 print('User deleted successfully.')224 else:225 print('Something went wrong. ' + str(msg))226 227 @app.command()228 @update_sqlite_path229 def update_user(email: str,230 password: Optional[str] = None,231 role: Annotated[Optional[bool],232 typer.Option("--admin/--nonadmin"233 )] = None,234 active: Annotated[Optional[bool],235 typer.Option("--active/--inactive"236 )] = None,237 console: Optional[bool] = True,238 json: Optional[bool] = False,239 sqlite_path: Optional[str] = None,240 ):241 """Update internal user."""242 243 data = dict()244 if password:245 if len(password) < 6:246 print("Password must be at least 6 characters long.")247 exit()248 # validate_password relies on the new password being present as249 # `newPassword` and `confirmPassword` in the data250 data['newPassword'] = password251 data['confirmPassword'] = password252 253 if role is not None:254 data['role'] = 1 if role else 2255 if active is not None:256 data['active'] = active257 258 app = create_app(config.APP_NAME + '-cli')259 with app.test_request_context():260 uid = ManageUsers.get_user(username=email,261 auth_source=INTERNAL)262 if not uid:263 print("User not found")264 else:265 status, msg = user_management_update_user(uid, data)266 if status:267 _user = ManageUsers.get_users_from_db(username=email,268 auth_source=INTERNAL,269 console=False)270 ManageUsers.display_user(_user[0], console, json)271 else:272 print('Something went wrong. ' + str(msg))273 274 @app.command()275 @update_sqlite_path276 def get_users(username: Optional[str] = None,277 auth_source: AuthType = None,278 json: Optional[bool] = False,279 sqlite_path: Optional[str] = None,280 ):281 282 ManageUsers.get_users_from_db(username, auth_source, True, json)283 284 def get_users_from_db(username: Optional[str] = None,285 auth_source: AuthType = None,286 console: Optional[bool] = True,287 json: Optional[bool] = False,288 ):289 """Get user(s) details."""290 291 app = create_app(config.APP_NAME + '-cli')292 with app.test_request_context():293 if username and auth_source:294 users = User.query.filter_by(username=username,295 auth_source=auth_source)296 elif not username and auth_source:297 users = User.query.filter_by(auth_source=auth_source)298 elif username and not auth_source:299 users = User.query.filter_by(username=username)300 else:301 users = User.query.all()302 users_data = []303 for u in users:304 _data = {'id': u.id,305 'username': u.username,306 'email': u.email,307 'active': u.active,308 'role': u.roles[0].id,309 'auth_source': u.auth_source,310 'locked': u.locked311 }312 users_data.append(_data)313 if console:314 ManageUsers.display_user(users_data, console, json)315 else:316 return users_data317 318 @app.command()319 @update_sqlite_path320 def update_external_user(username: str,321 auth_source: AuthExtTypes = AuthExtTypes.oauth2,322 email: Optional[str] = None,323 role: Annotated[Optional[bool],324 typer.Option("--admin/--nonadmin"325 )] = None,326 active: Annotated[327 Optional[bool],328 typer.Option("--active/--inactive")] = None,329 console: Optional[bool] = True,330 json: Optional[bool] = False,331 sqlite_path: Optional[str] = None,332 ):333 """Update external users other than Internal like334 Ldap, Ouath2, Kerberos, Webserver."""335 336 data = dict()337 if email:338 data['email'] = email339 if role is not None:340 data['role'] = 1 if role else 2341 if active is not None:342 data['active'] = active343 344 app = create_app(config.APP_NAME + '-cli')345 with app.test_request_context():346 uid = ManageUsers.get_user(username=username,347 auth_source=auth_source)348 if not uid:349 print("User not found")350 else:351 status, msg = user_management_update_user(uid, data)352 if status:353 _user = ManageUsers.get_users_from_db(354 username=username,355 auth_source=auth_source,356 console=False357 )358 ManageUsers.display_user(_user[0], console, json)359 else:360 print('Something went wrong. ' + str(msg))361 362 def create_user(data, console, json):363 app = create_app(config.APP_NAME + '-cli')364 with app.test_request_context():365 username = data['username'] if 'username' in data else \366 data['email']367 uid = ManageUsers.get_user(username=username,368 auth_source=data['auth_source'])369 if uid:370 print("User already exists.")371 exit()372 373 if 'newPassword' in data and len(data['newPassword']) < 6:374 print("Password must be at least 6 characters long.")375 exit()376 377 status, msg = create_user(data)378 if status:379 ManageUsers.display_user(data, console, json)380 else:381 print('Something went wrong. ' + str(msg))382 383 def get_user(username=None, auth_source=INTERNAL):384 app = create_app(config.APP_NAME + '-cli')385 usr = None386 with app.test_request_context():387 usr = User.query.filter_by(username=username,388 auth_source=auth_source).first()389 390 if not usr:391 return None392 return usr.id393 394 def display_user(data, _console, _json):395 if _console:396 if _json:397 json_formatted_str = jsonlib.dumps(data, indent=0)398 console.print(json_formatted_str)399 else:400 if isinstance(data, dict):401 data = [data]402 for _data in data:403 table = Table(title="User Details", box=box.ASCII)404 table.add_column("Field", style="green")405 table.add_column("Value", style="green")406 407 if 'username' in _data:408 table.add_row("Username", _data['username'])409 if 'email' in _data:410 table.add_row("Email", _data['email'])411 table.add_row("auth_source", _data['auth_source'])412 table.add_row("role",413 "Admin" if _data['role'] and414 _data['role'] != 2 else415 "Non-admin")416 table.add_row("active",417 'True' if _data['active'] else 'False')418 console.print(table)419 420 421class ManagePreferences:422 423 def get_user(username=None, auth_source=INTERNAL):424 app = create_app(config.APP_NAME + '-cli')425 usr = None426 with app.test_request_context():427 usr = User.query.filter_by(username=username,428 auth_source=auth_source).first()429 if not usr:430 return None431 return usr.id432 433 @app.command()434 @update_sqlite_path435 def get_prefs(json: Optional[bool] = False,436 sqlite_path: Optional[str] = None,437 ):438 439 return ManagePreferences.fetch_prefs()440 441 def fetch_prefs(id: Optional[bool] = None, json: Optional[bool] = False):442 """Get Preferences List."""443 app = create_app(config.APP_NAME + '-cli')444 table = Table(title="Pref Details", box=box.ASCII)445 table.add_column("Preference", style="green")446 with app.app_context():447 from pgadmin.model import Preferences as PrefTable, \448 ModulePreference as ModulePrefTable, \449 PreferenceCategory as PrefCategoryTbl450 451 module_prefs = ModulePrefTable.query.all()452 cat_prefs = PrefCategoryTbl.query.all()453 prefs = PrefTable.query.all()454 if id:455 all_preferences = {}456 else:457 all_preferences = []458 for i in module_prefs:459 for j in cat_prefs:460 if i.id == j.mid:461 for k in prefs:462 if k.cid == j.id:463 if id:464 all_preferences["{0}:{1}:{2}".format(465 i.name, j.name, k.name)466 ] = "{0}:{1}:{2}".format(i.id, j.id, k.id)467 else:468 table.add_row("{0}:{1}:{2}".format(469 i.name, j.name, k.name))470 all_preferences.append(471 "{0}:{1}:{2}".format(472 i.name, j.name, k.name)473 )474 if id:475 return all_preferences476 else:477 if json:478 json_formatted_str = jsonlib.dumps(479 {"Preferences": all_preferences},480 indent=0)481 print(json_formatted_str)482 else:483 print(table)484 485 @app.command()486 @update_sqlite_path487 def set_prefs(username,488 pref_options: Annotated[Optional[List[str]],489 typer.Argument()] = None,490 auth_source: AuthType = AuthType.internal,491 console: Optional[bool] = True,492 json: Optional[bool] = False,493 input_file: Optional[str] = None,494 sqlite_path: Optional[str] = None,495 ):496 """Set User preferences."""497 498 if input_file:499 from urllib.parse import unquote500 # generate full path of file501 try:502 file_path = unquote(input_file)503 except Exception as e:504 print(str(e))505 return _handle_error(str(e), True)506 import json as json_utility507 try:508 with open(file_path) as f:509 data = json_utility.load(f)510 except json_utility.decoder.JSONDecodeError as e:511 return _handle_error(gettext("Error parsing input file %s: %s"512 % (file_path, e)), True)513 except Exception as e:514 return _handle_error(515 gettext("Error reading input file %s: [%d] %s" %516 (file_path, e.errno, e.strerror)), True)517 518 pref_data = data['preferences']519 520 for k, v in pref_data.items():521 pref_options.append(k + "=" + str(v))522 523 user_id = ManagePreferences.get_user(username, auth_source)524 table = Table(title="Updated Pref Details", box=box.ASCII)525 table.add_column("Preference", style="green")526 if not user_id:527 print("User not found.")528 return529 530 prefs = ManagePreferences.fetch_prefs(True)531 app = create_app(config.APP_NAME + '-cli')532 invalid_prefs = []533 valid_prefs = []534 with app.app_context():535 from pgadmin.preferences import save_pref536 for opt in pref_options:537 val = opt.split("=")538 if len(val) <= 1:539 print('Preference key=value is required, example: '540 '[green]sqleditor:editor:comma_first=true[/green]')541 return542 final_opt = val[0].split(":")543 val = val[1]544 f = ":".join(final_opt)545 if f in prefs:546 ids = prefs[f].split(":")547 _row = {548 'mid': ids[0],549 'category_id': ids[1],550 'id': ids[2],551 'name': final_opt[2],552 'user_id': user_id,553 'value': val}554 save_pref(_row)555 valid_prefs.append(_row)556 557 if not json:558 table.add_row(jsonlib.dumps(_row))559 else:560 invalid_prefs.append(f)561 562 if len(invalid_prefs) >= 1:563 print("Preference(s) [red]{0}[/red] not found.".format(564 (', ').join(565 invalid_prefs)))566 567 if not json and console:568 print(table)569 elif json and console:570 print(jsonlib.dumps(valid_prefs, indent=2))571 572 573@app.command()574def setup_db(app: Annotated[str, typer.Argument(575 help="This argument doesn't require in CLI mode.")] = None):576 """Setup the configuration database."""577 578 app = app or create_app()579 create_app_data_directory(config)580 581 print("pgAdmin 4 - Application Initialisation")582 print("======================================\n")583 584 def run_migration_for_sqlite():585 with app.app_context():586 # Run migration for the first time i.e. create database587 from config import SQLITE_PATH588 if not os.path.exists(SQLITE_PATH):589 db_upgrade(app)590 else:591 version = Version.query.filter_by(name='ConfigDB').first()592 schema_version = version.value593 594 # Run migration if current schema version is greater than the595 # schema version stored in version table596 if CURRENT_SCHEMA_VERSION >= schema_version:597 db_upgrade(app)598 599 # Update schema version to the latest600 if CURRENT_SCHEMA_VERSION > schema_version:601 version = Version.query.filter_by(name='ConfigDB').first()602 version.value = CURRENT_SCHEMA_VERSION603 db.session.commit()604 605 if os.name != 'nt':606 os.chmod(config.SQLITE_PATH, 0o600)607 608 def run_migration_for_others():609 with app.app_context():610 version = Version.query.filter_by(name='ConfigDB').first()611 if version == -1:612 db_upgrade(app)613 else:614 schema_version = version.value615 616 # Run migration if current schema version is greater than the617 # schema version stored in version table618 if CURRENT_SCHEMA_VERSION >= schema_version:619 db_upgrade(app)620 621 # Update schema version to the latest622 if CURRENT_SCHEMA_VERSION > schema_version:623 version = Version.query.filter_by(name='ConfigDB').first()624 version.value = CURRENT_SCHEMA_VERSION625 db.session.commit()626 627 # Run the migration as per specified by the user.628 if config.CONFIG_DATABASE_URI is not None and \629 len(config.CONFIG_DATABASE_URI) > 0:630 run_migration_for_others()631 else:632 run_migration_for_sqlite()633 634 635def main():636 app()637 638 639if __name__ == "__main__":640 main()641