mirror of
https://github.com/MarioSpore/Grinch-AP.git
synced 2025-10-21 12:11:33 -06:00
SSL support (#1340)
This commit is contained in:
@@ -22,6 +22,9 @@ import ModuleUpdate
|
||||
|
||||
ModuleUpdate.update()
|
||||
|
||||
if typing.TYPE_CHECKING:
|
||||
import ssl
|
||||
|
||||
import websockets
|
||||
import colorama
|
||||
try:
|
||||
@@ -2090,6 +2093,8 @@ def parse_args() -> argparse.Namespace:
|
||||
parser.add_argument('--password', default=defaults["password"])
|
||||
parser.add_argument('--savefile', default=defaults["savefile"])
|
||||
parser.add_argument('--disable_save', default=defaults["disable_save"], action='store_true')
|
||||
parser.add_argument('--cert', help="Path to a SSL Certificate for encryption.")
|
||||
parser.add_argument('--cert_key', help="Path to SSL Certificate Key file")
|
||||
parser.add_argument('--loglevel', default=defaults["loglevel"],
|
||||
choices=['debug', 'info', 'warning', 'error', 'critical'])
|
||||
parser.add_argument('--location_check_points', default=defaults["location_check_points"], type=int)
|
||||
@@ -2162,6 +2167,14 @@ async def auto_shutdown(ctx, to_cancel=None):
|
||||
await asyncio.sleep(seconds)
|
||||
|
||||
|
||||
def load_server_cert(path: str, cert_key: typing.Optional[str]) -> "ssl.SSLContext":
|
||||
import ssl
|
||||
ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
|
||||
ssl_context.load_default_certs()
|
||||
ssl_context.load_cert_chain(path, cert_key if cert_key else path)
|
||||
return ssl_context
|
||||
|
||||
|
||||
async def main(args: argparse.Namespace):
|
||||
Utils.init_logging("Server", loglevel=args.loglevel.lower())
|
||||
|
||||
@@ -2197,8 +2210,10 @@ async def main(args: argparse.Namespace):
|
||||
|
||||
ctx.init_save(not args.disable_save)
|
||||
|
||||
ssl_context = load_server_cert(args.cert, args.cert_key) if args.cert else None
|
||||
|
||||
ctx.server = websockets.serve(functools.partial(server, ctx=ctx), host=ctx.host, port=ctx.port, ping_timeout=None,
|
||||
ping_interval=None)
|
||||
ping_interval=None, ssl=ssl_context)
|
||||
ip = args.host if args.host else Utils.get_public_ipv4()
|
||||
logging.info('Hosting game at %s:%d (%s)' % (ip, ctx.port,
|
||||
'No password' if not ctx.password else 'Password: %s' % ctx.password))
|
||||
|
Reference in New Issue
Block a user