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
|
# Copyright (C) 2013-2014 science + computing ag
# Author: Sebastian Deiss <sebastian.deiss@t-online.de>
#
#
# This file is part of paramiko.
#
# Paramiko is free software; you can redistribute it and/or modify it under the
# terms of the GNU Lesser General Public License as published by the Free
# Software Foundation; either version 2.1 of the License, or (at your option)
# any later version.
#
# Paramiko is distributed in the hope that it will be useful, but WITHOUT ANY
# WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR
# A PARTICULAR PURPOSE. See the GNU Lesser General Public License for more
# details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with Paramiko; if not, write to the Free Software Foundation, Inc.,
# 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA.
"""
Test the used APIs for GSS-API / SSPI authentication
@author: Sebastian Deiss
@contact: U{https://github.com/SebastianDeiss/paramiko/issues}
@organization: science + computing ag
(U{EMail<mailto:a.kruis@science-computing.de>})
@copyright: (C) 2013-2014 U{science + computing ag
<https://www.science-computing.de>}
@license: GNU Lesser General Public License (LGPL)
Created on 04.12.2013
"""
import unittest
import socket
class GSSAPITest(unittest.TestCase):
def init(hostname=None, srv_mode=False):
global krb5_mech, targ_name, server_mode
krb5_mech = "1.2.840.113554.1.2.2"
targ_name = hostname
server_mode = srv_mode
init = staticmethod(init)
def test_1_pyasn1(self):
"""
Test the used methods of pyasn1.
"""
from pyasn1.type.univ import ObjectIdentifier
from pyasn1.codec.der import encoder, decoder
oid = encoder.encode(ObjectIdentifier(krb5_mech))
mech, __ = decoder.decode(oid)
self.assertEquals(krb5_mech, mech.__str__())
def test_2_gssapi_sspi(self):
"""
Test the used methods of python-gssapi or sspi, sspicon from pywin32.
"""
_API = "MIT"
try:
import gssapi
except ImportError:
import sspicon
import sspi
_API = "SSPI"
c_token = None
gss_ctxt_status = False
mic_msg = b"G'day Mate!"
if _API == "MIT":
if server_mode:
gss_flags = (gssapi.C_PROT_READY_FLAG,
gssapi.C_INTEG_FLAG,
gssapi.C_MUTUAL_FLAG,
gssapi.C_DELEG_FLAG)
else:
gss_flags = (gssapi.C_PROT_READY_FLAG,
gssapi.C_INTEG_FLAG,
gssapi.C_DELEG_FLAG)
"""
Initialize a GSS-API context.
"""
ctx = gssapi.Context()
ctx.flags = gss_flags
krb5_oid = gssapi.OID.mech_from_string(krb5_mech)
target_name = gssapi.Name("host@" + targ_name,
gssapi.C_NT_HOSTBASED_SERVICE)
gss_ctxt = gssapi.InitContext(peer_name=target_name,
mech_type=krb5_oid,
req_flags=ctx.flags)
if server_mode:
c_token = gss_ctxt.step(c_token)
gss_ctxt_status = gss_ctxt.established
self.assertEquals(False, gss_ctxt_status)
"""
Accept a GSS-API context.
"""
gss_srv_ctxt = gssapi.AcceptContext()
s_token = gss_srv_ctxt.step(c_token)
gss_ctxt_status = gss_srv_ctxt.established
self.assertNotEquals(None, s_token)
self.assertEquals(True, gss_ctxt_status)
"""
Establish the client context
"""
c_token = gss_ctxt.step(s_token)
self.assertEquals(None, c_token)
else:
while not gss_ctxt.established:
c_token = gss_ctxt.step(c_token)
self.assertNotEquals(None, c_token)
"""
Build MIC
"""
mic_token = gss_ctxt.get_mic(mic_msg)
if server_mode:
"""
Check MIC
"""
status = gss_srv_ctxt.verify_mic(mic_msg, mic_token)
self.assertEquals(0, status)
else:
gss_flags = sspicon.ISC_REQ_INTEGRITY |\
sspicon.ISC_REQ_MUTUAL_AUTH |\
sspicon.ISC_REQ_DELEGATE
"""
Initialize a GSS-API context.
"""
target_name = "host/" + socket.getfqdn(targ_name)
gss_ctxt = sspi.ClientAuth("Kerberos",
scflags=gss_flags,
targetspn=target_name)
if server_mode:
error, token = gss_ctxt.authorize(c_token)
c_token = token[0].Buffer
self.assertEquals(0, error)
"""
Accept a GSS-API context.
"""
gss_srv_ctxt = sspi.ServerAuth("Kerberos", spn=target_name)
error, token = gss_srv_ctxt.authorize(c_token)
s_token = token[0].Buffer
"""
Establish the context.
"""
error, token = gss_ctxt.authorize(s_token)
c_token = token[0].Buffer
self.assertEquals(None, c_token)
self.assertEquals(0, error)
"""
Build MIC
"""
mic_token = gss_ctxt.sign(mic_msg)
"""
Check MIC
"""
gss_srv_ctxt.verify(mic_msg, mic_token)
else:
error, token = gss_ctxt.authorize(c_token)
c_token = token[0].Buffer
self.assertNotEquals(0, error)
|
