dbcli/mycli
1
0
Fork 0
Code Issues Activity

Merging upstream version 1.27.2.

Browse source 
Signed-off-by: Daniel Baumann <daniel@debian.org>
This commit is contained in:
Daniel Baumann 2025-02-09 19:13:22 +01:00
parent d3f72a1e51
commit cc1aa7d50e
Signed by: daniel
GPG key ID: FBB4F0E80A80222F
15 changed files with 248 additions and 118 deletions
Download patch file Download diff file Expand all files Collapse all files

13
mycli/AUTHORS
View file

@ -1,11 +1,7 @@
Project Lead:
-------------
* Thomas Roten
Core Developers:
----------------
* Thomas Roten
* Irina Truong
* Matheus Rosa
* Darik Gamble
@ -35,6 +31,7 @@ Contributors:
* Daniel Black
* Daniel West
* Daniël van Eeden
* Fabrizio Gennari
* François Pietka
* Frederic Aoustin
* Georgy Frolov
@ -94,6 +91,12 @@ Contributors:
* Arvind Mishra
* Kevin Schmeichel
* Mel Dafert
* Thomas Copper
* Will Wang
* Alfred Wingate
* Zhanze Wang
* Houston Wong
Created by:
-----------

2
mycli/__init__.py
View file

@ -1 +1 @@
__version__ = '1.27.0'
__version__ = '1.27.2'

5
mycli/clitoolbar.py
View file

@ -7,8 +7,7 @@ from .packages import special
def create_toolbar_tokens_func(mycli, show_fish_help):
"""Return a function that generates the toolbar tokens."""
def get_toolbar_tokens():
result = []
result.append(('class:bottom-toolbar', ' '))
result = [('class:bottom-toolbar', ' ')]
if mycli.multi_line:
delimiter = special.get_current_delimiter()
@ -26,7 +25,7 @@ def create_toolbar_tokens_func(mycli, show_fish_help):
'[F3] Multiline: OFF '))
if mycli.prompt_app.editing_mode == EditingMode.VI:
result.append((
'class:botton-toolbar.on',
'class:bottom-toolbar.on',
'Vi-mode ({})'.format(_get_vi_mode())
))

14
mycli/main.py
View file

@ -93,6 +93,7 @@ SUPPORT_INFO = (
class MyCli(object):
default_prompt = '\\t \\u@\\h:\\d> '
default_prompt_splitln = '\\u@\\h\\n(\\t):\\d>'
max_len_prompt = 45
defaults_suffix = None
@ -427,6 +428,7 @@ class MyCli(object):
port = 3306
if not host or host == 'localhost':
socket = (
socket or
cnf['socket'] or
cnf['default_socket'] or
guess_socket_location()
@ -643,7 +645,7 @@ class MyCli(object):
def get_message():
prompt = self.get_prompt(self.prompt_format)
if self.prompt_format == self.default_prompt and len(prompt) > self.max_len_prompt:
prompt = self.get_prompt('\\d> ')
prompt = self.get_prompt(self.default_prompt_splitln)
prompt = prompt.replace("\\x1b", "\x1b")
return ANSI(prompt)
@ -1135,6 +1137,9 @@ class MyCli(object):
@click.option('--ssl-key', help='X509 key in PEM format.',
type=click.Path(exists=True))
@click.option('--ssl-cipher', help='SSL cipher to use.')
@click.option('--tls-version',
type=click.Choice(['TLSv1', 'TLSv1.1', 'TLSv1.2', 'TLSv1.3'], case_sensitive=False),
help='TLS protocol version for secure connection.')
@click.option('--ssl-verify-server-cert', is_flag=True,
help=('Verify server\'s "Common Name" in its cert against '
'hostname used when connecting. This option is disabled '
@ -1186,8 +1191,8 @@ def cli(database, user, host, port, socket, password, dbname,
version, verbose, prompt, logfile, defaults_group_suffix,
defaults_file, login_path, auto_vertical_output, local_infile,
ssl_enable, ssl_ca, ssl_capath, ssl_cert, ssl_key, ssl_cipher,
ssl_verify_server_cert, table, csv, warn, execute, myclirc, dsn,
list_dsn, ssh_user, ssh_host, ssh_port, ssh_password,
tls_version, ssl_verify_server_cert, table, csv, warn, execute,
myclirc, dsn, list_dsn, ssh_user, ssh_host, ssh_port, ssh_password,
ssh_key_filename, list_ssh_config, ssh_config_path, ssh_config_host,
init_command, charset, password_file):
"""A MySQL terminal client with auto-completion and syntax highlighting.
@ -1246,6 +1251,7 @@ def cli(database, user, host, port, socket, password, dbname,
'key': ssl_key and os.path.expanduser(ssl_key),
'capath': ssl_capath,
'cipher': ssl_cipher,
'tls_version': tls_version,
'check_hostname': ssl_verify_server_cert,
}
@ -1278,7 +1284,7 @@ def cli(database, user, host, port, socket, password, dbname,
uri = urlparse(dsn_uri)
if not database:
database = uri.path[1:] # ignore the leading fwd slash
if not user:
if not user and uri.username is not None:
user = unquote(uri.username)
if not password and uri.password is not None:
password = unquote(uri.password)

43
mycli/sqlexecute.py
View file

@ -176,11 +176,15 @@ class SQLExecute(object):
if init_command and len(list(special.split_queries(init_command))) > 1:
client_flag |= pymysql.constants.CLIENT.MULTI_STATEMENTS
ssl_context = None
if ssl:
ssl_context = self._create_ssl_ctx(ssl)
conn = pymysql.connect(
database=db, user=user, password=password, host=host, port=port,
unix_socket=socket, use_unicode=True, charset=charset,
autocommit=True, client_flag=client_flag,
local_infile=local_infile, conv=conv, ssl=ssl, program_name="mycli",
local_infile=local_infile, conv=conv, ssl=ssl_context, program_name="mycli",
defer_connect=defer_connect, init_command=init_command
)
@ -354,3 +358,40 @@ class SQLExecute(object):
def change_db(self, db):
self.conn.select_db(db)
self.dbname = db
def _create_ssl_ctx(self, sslp):
import ssl
ca = sslp.get("ca")
capath = sslp.get("capath")
hasnoca = ca is None and capath is None
ctx = ssl.create_default_context(cafile=ca, capath=capath)
ctx.check_hostname = not hasnoca and sslp.get("check_hostname", True)
ctx.verify_mode = ssl.CERT_NONE if hasnoca else ssl.CERT_REQUIRED
if "cert" in sslp:
ctx.load_cert_chain(sslp["cert"], keyfile=sslp.get("key"))
if "cipher" in sslp:
ctx.set_ciphers(sslp["cipher"])
# raise this default to v1.1 or v1.2?
ctx.minimum_version = ssl.TLSVersion.TLSv1
if "tls_version" in sslp:
tls_version = sslp["tls_version"]
if tls_version == "TLSv1":
ctx.minimum_version = ssl.TLSVersion.TLSv1
ctx.maximum_version = ssl.TLSVersion.TLSv1
elif tls_version == "TLSv1.1":
ctx.minimum_version = ssl.TLSVersion.TLSv1_1
ctx.maximum_version = ssl.TLSVersion.TLSv1_1
elif tls_version == "TLSv1.2":
ctx.minimum_version = ssl.TLSVersion.TLSv1_2
ctx.maximum_version = ssl.TLSVersion.TLSv1_2
elif tls_version == "TLSv1.3":
ctx.minimum_version = ssl.TLSVersion.TLSv1_3
ctx.maximum_version = ssl.TLSVersion.TLSv1_3
else:
_logger.error('Invalid tls version: %s', tls_version)
return ctx