File: //proc/676643/root/usr/local/CyberPanel/lib64/python3.10/site-packages/asyncssh/gss_unix.py
# Copyright (c) 2017-2022 by Ron Frederick <ronf@timeheart.net> and others.
#
# This program and the accompanying materials are made available under
# the terms of the Eclipse Public License v2.0 which accompanies this
# distribution and is available at:
#
# http://www.eclipse.org/legal/epl-2.0/
#
# This program may also be made available under the following secondary
# licenses when the conditions for such availability set forth in the
# Eclipse Public License v2.0 are satisfied:
#
# GNU General Public License, Version 2.0, or any later versions of
# that license
#
# SPDX-License-Identifier: EPL-2.0 OR GPL-2.0-or-later
#
# Contributors:
# Ron Frederick - initial implementation, API, and documentation
"""GSSAPI wrapper for UNIX"""
from typing import Optional, Sequence, SupportsBytes, cast
from gssapi import Credentials, Name, NameType, OID
from gssapi import RequirementFlag, SecurityContext
from gssapi.exceptions import GSSError
from .asn1 import OBJECT_IDENTIFIER
from .misc import BytesOrStrDict
def _mech_to_oid(mech: OID) -> bytes:
"""Return a DER-encoded OID corresponding to the requested GSS mechanism"""
mech_bytes = bytes(cast(SupportsBytes, mech))
return bytes((OBJECT_IDENTIFIER, len(mech_bytes))) + mech_bytes
class GSSBase:
"""GSS base class"""
def __init__(self, host: str, store: Optional[BytesOrStrDict]):
if '@' in host:
self._host = Name(host)
else:
self._host = Name('host@' + host, NameType.hostbased_service)
self._store = store
self._mechs = [_mech_to_oid(mech) for mech in self._creds.mechs]
self._ctx: Optional[SecurityContext] = None
@property
def _creds(self) -> Credentials:
"""Abstract method to construct GSS credentials"""
raise NotImplementedError
def _init_context(self) -> None:
"""Abstract method to construct GSS security context"""
raise NotImplementedError
@property
def mechs(self) -> Sequence[bytes]:
"""Return GSS mechanisms available for this host"""
return self._mechs
@property
def complete(self) -> bool:
"""Return whether or not GSS negotiation is complete"""
return self._ctx.complete if self._ctx else False
@property
def provides_mutual_auth(self) -> bool:
"""Return whether or not this context provides mutual authentication"""
assert self._ctx is not None
return bool(self._ctx.actual_flags &
RequirementFlag.mutual_authentication)
@property
def provides_integrity(self) -> bool:
"""Return whether or not this context provides integrity protection"""
assert self._ctx is not None
return bool(self._ctx.actual_flags & RequirementFlag.integrity)
@property
def user(self) -> str:
"""Return user principal associated with this context"""
assert self._ctx is not None
return str(self._ctx.initiator_name)
@property
def host(self) -> str:
"""Return host principal associated with this context"""
assert self._ctx is not None
return str(self._ctx.target_name)
def reset(self) -> None:
"""Reset GSS security context"""
self._ctx = None
def step(self, token: Optional[bytes] = None) -> Optional[bytes]:
"""Perform next step in GSS security exchange"""
if not self._ctx:
self._init_context()
assert self._ctx is not None
return self._ctx.step(token)
def sign(self, data: bytes) -> bytes:
"""Sign a block of data"""
assert self._ctx is not None
return self._ctx.get_signature(data)
def verify(self, data: bytes, sig: bytes) -> bool:
"""Verify a signature for a block of data"""
assert self._ctx is not None
try:
self._ctx.verify_signature(data, sig)
return True
except GSSError:
return False
class GSSClient(GSSBase):
"""GSS client"""
def __init__(self, host: str, store: Optional[BytesOrStrDict],
delegate_creds: bool):
super().__init__(host, store)
flags = RequirementFlag.mutual_authentication | \
RequirementFlag.integrity
if delegate_creds:
flags |= RequirementFlag.delegate_to_peer
self._flags = flags
@property
def _creds(self) -> Credentials:
"""Abstract method to construct GSS credentials"""
return Credentials(usage='initiate', store=self._store)
def _init_context(self) -> None:
"""Construct GSS client security context"""
self._ctx = SecurityContext(name=self._host, creds=self._creds,
flags=self._flags)
class GSSServer(GSSBase):
"""GSS server"""
@property
def _creds(self) -> Credentials:
"""Abstract method to construct GSS credentials"""
return Credentials(name=self._host, usage='accept', store=self._store)
def _init_context(self) -> None:
"""Construct GSS server security context"""
self._ctx = SecurityContext(creds=self._creds)