Jocay
288 lines
9.3 KiB
Python
Executable File
288 lines
9.3 KiB
Python
Executable File
import logging
|
|
import os.path
|
|
import secrets
|
|
import ssl
|
|
import sys
|
|
|
|
from tornado.options import define
|
|
from webssh.auth import AuthManager
|
|
from webssh.policy import (
|
|
load_host_keys, get_policy_class, check_policy_setting
|
|
)
|
|
from webssh.storage import ConnectionStore
|
|
from webssh.utils import (
|
|
to_ip_address, parse_origin_from_url, is_valid_encoding
|
|
)
|
|
from webssh._version import __version__
|
|
|
|
|
|
def print_version(flag):
|
|
if flag:
|
|
print(__version__)
|
|
sys.exit(0)
|
|
|
|
|
|
define('address', default='', help='Listen address')
|
|
define('port', type=int, default=8888, help='Listen port')
|
|
define('ssladdress', default='', help='SSL listen address')
|
|
define('sslport', type=int, default=4433, help='SSL listen port')
|
|
define('certfile', default='', help='SSL certificate file')
|
|
define('keyfile', default='', help='SSL private key file')
|
|
define('debug', type=bool, default=False, help='Debug mode')
|
|
define('policy', default='warning',
|
|
help='Missing host key policy, reject|autoadd|warning')
|
|
define('hostfile', default='', help='User defined host keys file')
|
|
define('syshostfile', default='', help='System wide host keys file')
|
|
define('tdstream', default='', help='Trusted downstream, separated by comma')
|
|
define('redirect', type=bool, default=True, help='Redirecting http to https')
|
|
define('fbidhttp', type=bool, default=True,
|
|
help='Forbid public plain http incoming requests')
|
|
define('xheaders', type=bool, default=True, help='Support xheaders')
|
|
define('xsrf', type=bool, default=True, help='CSRF protection')
|
|
define('origin', default='same', help='''Origin policy,
|
|
'same': same origin policy, matches host name and port number;
|
|
'primary': primary domain policy, matches primary domain only;
|
|
'<domains>': custom domains policy, matches any domain in the <domains> list
|
|
separated by comma;
|
|
'*': wildcard policy, matches any domain, allowed in debug mode only.''')
|
|
define('wpintvl', type=float, default=30, help='Websocket ping interval')
|
|
define('timeout', type=float, default=3, help='SSH connection timeout')
|
|
define('delay', type=float, default=3, help='The delay to call recycle_worker')
|
|
define('maxconn', type=int, default=20,
|
|
help='Maximum live connections (ssh sessions) per client')
|
|
define('font', default='', help='custom font filename')
|
|
define('encoding', default='',
|
|
help='''The default character encoding of ssh servers.
|
|
Example: --encoding='utf-8' to solve the problem with some switches&routers''')
|
|
define('auth', type=bool, default=True,
|
|
help='Require a WebSSH login before connecting to ssh servers')
|
|
define('auth_username', default=os.environ.get('WEBSSH_AUTH_USERNAME',
|
|
'admin'),
|
|
help='WebSSH login username used before first-time setup')
|
|
define('auth_password', default=os.environ.get('WEBSSH_AUTH_PASSWORD', ''),
|
|
help='WebSSH login password, preferably supplied by environment')
|
|
define('auth_password_hash',
|
|
default=os.environ.get('WEBSSH_AUTH_PASSWORD_HASH', ''),
|
|
help='PBKDF2 password hash for WebSSH login')
|
|
define('auth_session_days', type=int, default=7,
|
|
help='Days before the WebSSH login session expires')
|
|
define('data_dir',
|
|
default=os.environ.get(
|
|
'WEBSSH_DATA_DIR',
|
|
os.path.join(os.path.expanduser('~'), '.webssh')
|
|
),
|
|
help='Directory used for WebSSH auth and saved connection data')
|
|
define('version', type=bool, help='Show version information',
|
|
callback=print_version)
|
|
|
|
|
|
base_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|
font_dirs = ['webssh', 'static', 'css', 'fonts']
|
|
max_body_size = 1 * 1024 * 1024
|
|
|
|
|
|
class Font(object):
|
|
|
|
def __init__(self, filename, dirs):
|
|
self.family = self.get_family(filename)
|
|
self.url = self.get_url(filename, dirs)
|
|
# path relative to static_path, for static_url() in templates
|
|
self.path = self.get_url(filename, dirs[1:])
|
|
|
|
def get_family(self, filename):
|
|
return filename.split('.')[0]
|
|
|
|
def get_url(self, filename, dirs):
|
|
return '/'.join(dirs + [filename])
|
|
|
|
|
|
def get_app_settings(options):
|
|
auth_manager = get_auth_manager(options)
|
|
secret = get_cookie_secret(options)
|
|
settings = dict(
|
|
template_path=os.path.join(base_dir, 'webssh', 'templates'),
|
|
static_path=os.path.join(base_dir, 'webssh', 'static'),
|
|
websocket_ping_interval=options.wpintvl,
|
|
debug=options.debug,
|
|
xsrf_cookies=options.xsrf,
|
|
login_url='/login',
|
|
auth_manager=auth_manager,
|
|
auth_session_days=options.auth_session_days,
|
|
connection_store=get_connection_store(options, secret),
|
|
font=Font(
|
|
get_font_filename(options.font,
|
|
os.path.join(base_dir, *font_dirs)),
|
|
font_dirs[1:]
|
|
),
|
|
origin_policy=get_origin_setting(options)
|
|
)
|
|
if auth_manager:
|
|
settings['cookie_secret'] = secret
|
|
return settings
|
|
|
|
|
|
def get_server_settings(options):
|
|
settings = dict(
|
|
xheaders=options.xheaders,
|
|
max_body_size=max_body_size,
|
|
trusted_downstream=get_trusted_downstream(options.tdstream)
|
|
)
|
|
return settings
|
|
|
|
|
|
def get_host_keys_settings(options):
|
|
if not options.hostfile:
|
|
host_keys_filename = os.path.join(base_dir, 'known_hosts')
|
|
else:
|
|
host_keys_filename = options.hostfile
|
|
host_keys = load_host_keys(host_keys_filename)
|
|
|
|
if not options.syshostfile:
|
|
filename = os.path.expanduser('~/.ssh/known_hosts')
|
|
else:
|
|
filename = options.syshostfile
|
|
system_host_keys = load_host_keys(filename)
|
|
|
|
settings = dict(
|
|
host_keys=host_keys,
|
|
system_host_keys=system_host_keys,
|
|
host_keys_filename=host_keys_filename
|
|
)
|
|
return settings
|
|
|
|
|
|
def get_policy_setting(options, host_keys_settings):
|
|
policy_class = get_policy_class(options.policy)
|
|
logging.info(policy_class.__name__)
|
|
check_policy_setting(policy_class, host_keys_settings)
|
|
return policy_class()
|
|
|
|
|
|
def get_ssl_context(options):
|
|
if not options.certfile and not options.keyfile:
|
|
return None
|
|
elif not options.certfile:
|
|
raise ValueError('certfile is not provided')
|
|
elif not options.keyfile:
|
|
raise ValueError('keyfile is not provided')
|
|
elif not os.path.isfile(options.certfile):
|
|
raise ValueError('File {!r} does not exist'.format(options.certfile))
|
|
elif not os.path.isfile(options.keyfile):
|
|
raise ValueError('File {!r} does not exist'.format(options.keyfile))
|
|
else:
|
|
ssl_ctx = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
|
|
ssl_ctx.load_cert_chain(options.certfile, options.keyfile)
|
|
return ssl_ctx
|
|
|
|
|
|
def get_trusted_downstream(tdstream):
|
|
result = set()
|
|
for ip in tdstream.split(','):
|
|
ip = ip.strip()
|
|
if ip:
|
|
to_ip_address(ip)
|
|
result.add(ip)
|
|
return result
|
|
|
|
|
|
def get_origin_setting(options):
|
|
if options.origin == '*':
|
|
if not options.debug:
|
|
raise ValueError(
|
|
'Wildcard origin policy is only allowed in debug mode.'
|
|
)
|
|
else:
|
|
return '*'
|
|
|
|
origin = options.origin.lower()
|
|
if origin in ['same', 'primary']:
|
|
return origin
|
|
|
|
origins = set()
|
|
for url in origin.split(','):
|
|
orig = parse_origin_from_url(url)
|
|
if orig:
|
|
origins.add(orig)
|
|
|
|
if not origins:
|
|
raise ValueError('Empty origin list')
|
|
|
|
return origins
|
|
|
|
|
|
def get_font_filename(font, font_dir):
|
|
filenames = {f for f in os.listdir(font_dir) if not f.startswith('.')
|
|
and os.path.isfile(os.path.join(font_dir, f))}
|
|
if font:
|
|
if font not in filenames:
|
|
raise ValueError(
|
|
'Font file {!r} not found'.format(os.path.join(font_dir, font))
|
|
)
|
|
elif filenames:
|
|
font = filenames.pop()
|
|
|
|
return font
|
|
|
|
|
|
def check_encoding_setting(encoding):
|
|
if encoding and not is_valid_encoding(encoding):
|
|
raise ValueError('Unknown character encoding {!r}.'.format(encoding))
|
|
|
|
|
|
def get_auth_manager(options):
|
|
if not options.auth:
|
|
return
|
|
|
|
auth_file = os.path.join(
|
|
os.path.expanduser(options.data_dir), 'auth.json'
|
|
)
|
|
return AuthManager(
|
|
username=options.auth_username,
|
|
password=options.auth_password,
|
|
password_hash=options.auth_password_hash,
|
|
auth_file=auth_file
|
|
)
|
|
|
|
|
|
def get_connection_store(options, secret=''):
|
|
filename = os.path.join(
|
|
os.path.expanduser(options.data_dir), 'connections.json'
|
|
)
|
|
store = ConnectionStore(filename, secret)
|
|
encrypted = store.encrypt_plaintext_passwords()
|
|
if encrypted:
|
|
logging.info(
|
|
'Encrypted {} plaintext password(s) in {}'.format(
|
|
encrypted, filename
|
|
)
|
|
)
|
|
return store
|
|
|
|
|
|
def get_cookie_secret(options):
|
|
secret = os.environ.get('WEBSSH_COOKIE_SECRET', '')
|
|
if secret:
|
|
return secret
|
|
|
|
filename = os.path.join(
|
|
os.path.expanduser(options.data_dir), 'cookie_secret'
|
|
)
|
|
if os.path.isfile(filename):
|
|
with open(filename, encoding='utf-8') as f:
|
|
secret = f.read().strip()
|
|
if secret:
|
|
return secret
|
|
|
|
dirname = os.path.dirname(filename)
|
|
if dirname and not os.path.isdir(dirname):
|
|
os.makedirs(dirname)
|
|
|
|
secret = secrets.token_urlsafe(64)
|
|
with open(filename, 'w', encoding='utf-8') as f:
|
|
f.write(secret + '\n')
|
|
try:
|
|
os.chmod(filename, 0o600)
|
|
except OSError:
|
|
pass
|
|
return secret
|