1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
|
# 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)
|
