python/samba/getopt.py

   1 # Samba-specific bits for optparse
   2 # Copyright (C) Jelmer Vernooij <jelmer@samba.org> 2007
   3 #
   4 # This program is free software; you can redistribute it and/or modify
   5 # it under the terms of the GNU General Public License as published by
   6 # the Free Software Foundation; either version 3 of the License, or
   7 # (at your option) any later version.
   8 #
   9 # This program is distributed in the hope that it will be useful,
  10 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  11 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  12 # GNU General Public License for more details.
  13 #
  14 # You should have received a copy of the GNU General Public License
  15 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
  16 #
  17
  18 """Support for parsing Samba-related command-line options."""
  19
  20 __docformat__ = "restructuredText"
  21
  22 import optparse
  23 import os
  24 import sys
  25 from abc import ABCMeta, abstractmethod
  26 from copy import copy
  27
  28 from samba.credentials import (
  29     Credentials,
  30     AUTO_USE_KERBEROS,
  31     DONT_USE_KERBEROS,
  32     MUST_USE_KERBEROS,
  33 )
  34 from samba._glue import get_burnt_commandline
  35
  36
  37 def check_bytes(option, opt, value):
  38     """Custom option type to allow the input of sizes using byte, kb, mb ...
  39
  40     units, e.g. 2Gb, 4KiB ...
  41        e.g. Option("--size", type="bytes", metavar="SIZE")
  42     """
  43
  44     multipliers = {"B": 1,
  45                    "KB": 1024,
  46                    "MB": 1024 * 1024,
  47                    "GB": 1024 * 1024 * 1024}
  48
  49     # strip out any spaces
  50     v = value.replace(" ", "")
  51
  52     # extract the numeric prefix
  53     digits = ""
  54     while v and v[0:1].isdigit() or v[0:1] == '.':
  55         digits += v[0]
  56         v = v[1:]
  57
  58     try:
  59         m = float(digits)
  60     except ValueError:
  61         msg = ("{0} option requires a numeric value, "
  62                "with an optional unit suffix").format(opt)
  63         raise optparse.OptionValueError(msg)
  64
  65     # strip out the 'i' and convert to upper case so
  66     # kib Kib kb KB are all equivalent
  67     suffix = v.upper().replace("I", "")
  68     try:
  69         return m * multipliers[suffix]
  70     except KeyError as k:
  71         msg = ("{0} invalid suffix '{1}', "
  72                "should be B, Kb, Mb or Gb").format(opt, v)
  73         raise optparse.OptionValueError(msg)
  74
  75
  76 class OptionMissingError(optparse.OptionValueError):
  77     """One or more Options with required=True is missing."""
  78
  79     def __init__(self, options):
  80         """Raised when required Options are missing from the command line.
  81
  82         :param options: list of 1 or more option
  83         """
  84         self.options = options
  85
  86     def __str__(self):
  87         if len(self.options) == 1:
  88             missing = self.options[0]
  89             return f"Argument {missing} is required."
  90         else:
  91             options = sorted([str(option) for option in self.options])
  92             missing = ", ".join(options)
  93             return f"The arguments {missing} are required."
  94
  95
  96 class ValidationError(Exception):
  97     """ValidationError is the exception raised by validators.
  98
  99     Should be raised from the __call__ method of the Validator subclass.
 100     """
 101     pass
 102
 103
 104 class Validator(metaclass=ABCMeta):
 105     """Base class for Validators used by SambaOption.
 106
 107     Subclass this to make custom validators and implement __call__.
 108     """
 109
 110     @abstractmethod
 111     def __call__(self, field, value):
 112         pass
 113
 114
 115 class Option(optparse.Option):
 116     ATTRS = optparse.Option.ATTRS + ["required", "validators"]
 117     TYPES = optparse.Option.TYPES + ("bytes",)
 118     TYPE_CHECKER = copy(optparse.Option.TYPE_CHECKER)
 119     TYPE_CHECKER["bytes"] = check_bytes
 120
 121     def run_validators(self, opt, value):
 122         """Runs the list of validators on the current option."""
 123         validators = getattr(self, "validators") or []
 124         for validator in validators:
 125             validator(opt, value)
 126
 127     def convert_value(self, opt, value):
 128         """Override convert_value to run validators just after.
 129
 130         This can also be done in process() but there we would have to
 131         replace the entire method.
 132         """
 133         value = super().convert_value(opt, value)
 134         self.run_validators(opt, value)
 135         return value
 136
 137
 138 class OptionParser(optparse.OptionParser):
 139     """Samba OptionParser, adding support for required=True on Options."""
 140
 141     def __init__(self,
 142                  usage=None,
 143                  option_list=None,
 144                  option_class=Option,
 145                  version=None,
 146                  conflict_handler="error",
 147                  description=None,
 148                  formatter=None,
 149                  add_help_option=True,
 150                  prog=None,
 151                  epilog=None):
 152         """
 153         Ensure that option_class defaults to the Samba one.
 154         """
 155         super().__init__(usage, option_list, option_class, version,
 156                          conflict_handler, description, formatter,
 157                          add_help_option, prog, epilog)
 158
 159     def check_values(self, values, args):
 160         """Loop through required options if value is missing raise exception."""
 161         missing = []
 162         for option in self._get_all_options():
 163             if option.required:
 164                 value = getattr(values, option.dest)
 165                 if value is None:
 166                     missing.append(option)
 167
 168         if missing:
 169             raise OptionMissingError(missing)
 170
 171         return super().check_values(values, args)
 172
 173
 174 class SambaOptions(optparse.OptionGroup):
 175     """General Samba-related command line options."""
 176
 177     def __init__(self, parser):
 178         from samba import fault_setup
 179         fault_setup()
 180
 181         # This removes passwords from the commandline via
 182         # setproctitle() but makes no change to python sys.argv so we
 183         # can continue to process as normal
 184         #
 185         # get_burnt_commandline returns None if no change is needed
 186         new_proctitle = get_burnt_commandline(sys.argv)
 187         if new_proctitle is not None:
 188             try:
 189                 import setproctitle
 190                 setproctitle.setproctitle(new_proctitle)
 191
 192             except ModuleNotFoundError:
 193                 msg = ("WARNING: Using passwords on command line is insecure. "
 194                        "Installing the setproctitle python module will hide "
 195                        "these from shortly after program start.\n")
 196                 sys.stderr.write(msg)
 197                 sys.stderr.flush()
 198
 199         from samba.param import LoadParm
 200         super().__init__(parser, "Samba Common Options")
 201         self.add_option("-s", "--configfile", action="callback",
 202                         type=str, metavar="FILE", help="Configuration file",
 203                         callback=self._load_configfile)
 204         self.add_option("-d", "--debuglevel", action="callback",
 205                         type=str, metavar="DEBUGLEVEL", help="debug level",
 206                         callback=self._set_debuglevel)
 207         self.add_option("--option", action="callback",
 208                         type=str, metavar="OPTION",
 209                         help="set smb.conf option from command line",
 210                         callback=self._set_option)
 211         self.add_option("--realm", action="callback",
 212                         type=str, metavar="REALM", help="set the realm name",
 213                         callback=self._set_realm)
 214         self._configfile = None
 215         self._lp = LoadParm()
 216         self.realm = None
 217
 218     def get_loadparm_path(self):
 219         """Return path to the smb.conf file specified on the command line."""
 220         return self._configfile
 221
 222     def _load_configfile(self, option, opt_str, arg, parser):
 223         self._configfile = arg
 224
 225     def _set_debuglevel(self, option, opt_str, arg, parser):
 226         try:
 227             self._lp.set('debug level', arg)
 228         except RuntimeError:
 229             raise optparse.OptionValueError(
 230                 f"invalid -d/--debug value: '{arg}'")
 231         parser.values.debuglevel = arg
 232
 233     def _set_realm(self, option, opt_str, arg, parser):
 234         try:
 235             self._lp.set('realm', arg)
 236         except RuntimeError:
 237             raise optparse.OptionValueError(
 238                 f"invalid --realm value: '{arg}'")
 239         self.realm = arg
 240
 241     def _set_option(self, option, opt_str, arg, parser):
 242         if arg.find('=') == -1:
 243             raise optparse.OptionValueError(
 244                 "--option option takes a 'a=b' argument")
 245         a = arg.split('=', 1)
 246         try:
 247             self._lp.set(a[0], a[1])
 248         except Exception as e:
 249             raise optparse.OptionValueError(
 250                 "invalid --option option value %r: %s" % (arg, e))
 251
 252     def get_loadparm(self):
 253         """Return loadparm object with data specified on the command line."""
 254         if self._configfile is not None:
 255             self._lp.load(self._configfile)
 256         elif os.getenv("SMB_CONF_PATH") is not None:
 257             self._lp.load(os.getenv("SMB_CONF_PATH"))
 258         else:
 259             self._lp.load_default()
 260         return self._lp
 261
 262
 263 class Samba3Options(SambaOptions):
 264     """General Samba-related command line options with an s3 param."""
 265
 266     def __init__(self, parser):
 267         super().__init__(parser)
 268         from samba.samba3 import param as s3param
 269         self._lp = s3param.get_context()
 270
 271
 272 class HostOptions(optparse.OptionGroup):
 273     """Command line options for connecting to target host or database."""
 274
 275     def __init__(self, parser):
 276         super().__init__(parser, "Host Options")
 277
 278         self.H = None
 279         self.add_option("-H", "--URL",
 280                         help="LDB URL for database or target server",
 281                         type=str, metavar="URL", action="callback",
 282                         callback=self._set_H, dest="H")
 283
 284     def _set_H(self, option, opt_str, arg, parser):
 285         parser.values.H = self.H = arg
 286
 287
 288 class VersionOptions(optparse.OptionGroup):
 289     """Command line option for printing Samba version."""
 290     def __init__(self, parser):
 291         super().__init__(parser, "Version Options")
 292         self.add_option("-V", "--version", action="callback",
 293                         callback=self._display_version,
 294                         help="Display version number")
 295
 296     def _display_version(self, option, opt_str, arg, parser):
 297         import samba
 298         print(samba.version)
 299         sys.exit(0)
 300
 301
 302 def parse_kerberos_arg_legacy(arg, opt_str):
 303     if arg.lower() in ["yes", 'true', '1']:
 304         return MUST_USE_KERBEROS
 305     elif arg.lower() in ["no", 'false', '0']:
 306         return DONT_USE_KERBEROS
 307     elif arg.lower() in ["auto"]:
 308         return AUTO_USE_KERBEROS
 309     else:
 310         raise optparse.OptionValueError("invalid %s option value: %s" %
 311                                         (opt_str, arg))
 312
 313
 314 def parse_kerberos_arg(arg, opt_str):
 315     if arg.lower() == 'required':
 316         return MUST_USE_KERBEROS
 317     elif arg.lower() == 'desired':
 318         return AUTO_USE_KERBEROS
 319     elif arg.lower() == 'off':
 320         return DONT_USE_KERBEROS
 321     else:
 322         raise optparse.OptionValueError("invalid %s option value: %s" %
 323                                         (opt_str, arg))
 324
 325
 326 class CredentialsOptions(optparse.OptionGroup):
 327     """Command line options for specifying credentials."""
 328
 329     def __init__(self, parser, special_name=None):
 330         self.special_name = special_name
 331         if special_name is not None:
 332             self.section = "Credentials Options (%s)" % special_name
 333         else:
 334             self.section = "Credentials Options"
 335
 336         self.ask_for_password = True
 337         self.ipaddress = None
 338         self.machine_pass = False
 339         super().__init__(parser, self.section)
 340         self._add_option("--simple-bind-dn", metavar="DN", action="callback",
 341                          callback=self._set_simple_bind_dn, type=str,
 342                          help="DN to use for a simple bind")
 343         self._add_option("--password", metavar="PASSWORD", action="callback",
 344                          help="Password", type=str, callback=self._set_password)
 345         self._add_option("-U", "--username", metavar="USERNAME",
 346                          action="callback", type=str,
 347                          help="Username", callback=self._parse_username)
 348         self._add_option("-W", "--workgroup", metavar="WORKGROUP",
 349                          action="callback", type=str,
 350                          help="Workgroup", callback=self._parse_workgroup)
 351         self._add_option("-N", "--no-pass", action="callback",
 352                          help="Don't ask for a password",
 353                          callback=self._set_no_password)
 354         self._add_option("", "--ipaddress", metavar="IPADDRESS",
 355                          action="callback", type=str,
 356                          help="IP address of server",
 357                          callback=self._set_ipaddress)
 358         self._add_option("-P", "--machine-pass",
 359                          action="callback",
 360                          help="Use stored machine account password",
 361                          callback=self._set_machine_pass)
 362         self._add_option("--use-kerberos", metavar="desired|required|off",
 363                          action="callback", type=str,
 364                          help="Use Kerberos authentication", callback=self._set_kerberos)
 365         self._add_option("--use-krb5-ccache", metavar="KRB5CCNAME",
 366                          action="callback", type=str,
 367                          help="Kerberos Credentials cache",
 368                          callback=self._set_krb5_ccache)
 369         self._add_option("-A", "--authentication-file", metavar="AUTHFILE",
 370                          action="callback", type=str,
 371                          help="Authentication file",
 372                          callback=self._set_auth_file)
 373
 374         # LEGACY
 375         self._add_option("-k", "--kerberos", metavar="KERBEROS",
 376                          action="callback", type=str,
 377                          help="DEPRECATED: Migrate to --use-kerberos", callback=self._set_kerberos_legacy)
 378         self.creds = Credentials()
 379
 380     def _add_option(self, *args1, **kwargs):
 381         if self.special_name is None:
 382             return self.add_option(*args1, **kwargs)
 383
 384         args2 = ()
 385         for a in args1:
 386             if not a.startswith("--"):
 387                 continue
 388             args2 += (a.replace("--", "--%s-" % self.special_name),)
 389         self.add_option(*args2, **kwargs)
 390
 391     def _parse_username(self, option, opt_str, arg, parser):
 392         self.creds.parse_string(arg)
 393         self.machine_pass = False
 394
 395     def _parse_workgroup(self, option, opt_str, arg, parser):
 396         self.creds.set_domain(arg)
 397
 398     def _set_password(self, option, opt_str, arg, parser):
 399         self.creds.set_password(arg)
 400         self.ask_for_password = False
 401         self.machine_pass = False
 402
 403     def _set_no_password(self, option, opt_str, arg, parser):
 404         self.ask_for_password = False
 405
 406     def _set_machine_pass(self, option, opt_str, arg, parser):
 407         self.machine_pass = True
 408
 409     def _set_ipaddress(self, option, opt_str, arg, parser):
 410         self.ipaddress = arg
 411
 412     def _set_kerberos_legacy(self, option, opt_str, arg, parser):
 413         print('WARNING: The option -k|--kerberos is deprecated!')
 414         self.creds.set_kerberos_state(parse_kerberos_arg_legacy(arg, opt_str))
 415
 416     def _set_kerberos(self, option, opt_str, arg, parser):
 417         self.creds.set_kerberos_state(parse_kerberos_arg(arg, opt_str))
 418
 419     def _set_simple_bind_dn(self, option, opt_str, arg, parser):
 420         self.creds.set_bind_dn(arg)
 421
 422     def _set_krb5_ccache(self, option, opt_str, arg, parser):
 423         self.creds.set_kerberos_state(MUST_USE_KERBEROS)
 424         self.creds.set_named_ccache(arg)
 425
 426     def _set_auth_file(self, option, opt_str, arg, parser):
 427         if os.path.exists(arg):
 428             self.creds.parse_file(arg)
 429             self.ask_for_password = False
 430             self.machine_pass = False
 431
 432     def get_credentials(self, lp, fallback_machine=False):
 433         """Obtain the credentials set on the command-line.
 434
 435         :param lp: Loadparm object to use.
 436         :return: Credentials object
 437         """
 438         self.creds.guess(lp)
 439         if self.machine_pass:
 440             self.creds.set_machine_account(lp)
 441         elif self.ask_for_password:
 442             self.creds.set_cmdline_callbacks()
 443
 444         # possibly fallback to using the machine account, if we have
 445         # access to the secrets db
 446         if fallback_machine and not self.creds.authentication_requested():
 447             try:
 448                 self.creds.set_machine_account(lp)
 449             except Exception:
 450                 pass
 451
 452         return self.creds
 453
 454
 455 class CredentialsOptionsDouble(CredentialsOptions):
 456     """Command line options for specifying credentials of two servers."""
 457
 458     def __init__(self, parser):
 459         super().__init__(parser)
 460         self.no_pass2 = True
 461         self.add_option("--simple-bind-dn2", metavar="DN2", action="callback",
 462                         callback=self._set_simple_bind_dn2, type=str,
 463                         help="DN to use for a simple bind")
 464         self.add_option("--password2", metavar="PASSWORD2", action="callback",
 465                         help="Password", type=str,
 466                         callback=self._set_password2)
 467         self.add_option("--username2", metavar="USERNAME2",
 468                         action="callback", type=str,
 469                         help="Username for second server",
 470                         callback=self._parse_username2)
 471         self.add_option("--workgroup2", metavar="WORKGROUP2",
 472                         action="callback", type=str,
 473                         help="Workgroup for second server",
 474                         callback=self._parse_workgroup2)
 475         self.add_option("--no-pass2", action="store_true",
 476                         help="Don't ask for a password for the second server")
 477         self.add_option("--use-kerberos2", metavar="desired|required|off",
 478                         action="callback", type=str,
 479                         help="Use Kerberos authentication", callback=self._set_kerberos2)
 480
 481         # LEGACY
 482         self.add_option("--kerberos2", metavar="KERBEROS2",
 483                         action="callback", type=str,
 484                         help="Use Kerberos", callback=self._set_kerberos2_legacy)
 485         self.creds2 = Credentials()
 486
 487     def _parse_username2(self, option, opt_str, arg, parser):
 488         self.creds2.parse_string(arg)
 489
 490     def _parse_workgroup2(self, option, opt_str, arg, parser):
 491         self.creds2.set_domain(arg)
 492
 493     def _set_password2(self, option, opt_str, arg, parser):
 494         self.creds2.set_password(arg)
 495         self.no_pass2 = False
 496
 497     def _set_kerberos2_legacy(self, option, opt_str, arg, parser):
 498         self.creds2.set_kerberos_state(parse_kerberos_arg(arg, opt_str))
 499
 500     def _set_kerberos2(self, option, opt_str, arg, parser):
 501         self.creds2.set_kerberos_state(parse_kerberos_arg(arg, opt_str))
 502
 503     def _set_simple_bind_dn2(self, option, opt_str, arg, parser):
 504         self.creds2.set_bind_dn(arg)
 505
 506     def get_credentials2(self, lp, guess=True):
 507         """Obtain the credentials set on the command-line.
 508
 509         :param lp: Loadparm object to use.
 510         :param guess: Try guess Credentials from environment
 511         :return: Credentials object
 512         """
 513         if guess:
 514             self.creds2.guess(lp)
 515         elif not self.creds2.get_username():
 516             self.creds2.set_anonymous()
 517
 518         if self.no_pass2:
 519             self.creds2.set_cmdline_callbacks()
 520         return self.creds2