openvpn-2fa-otp-freeradius-ldap

1
#! /usr/bin/env python3
2

3
import radiusd
4
import yaml
5
from typing import Tuple, TypedDict, List, Literal
6
import pyotp
7
from strongpass_ldap import LDAPConnection
8
from strongpass_db import DBConnection
9
from config import Config, CONFIG_FILE
10
from base64 import b64decode
11
from util import log_msg
12

13
cfg: Config
14
ldap_conn: LDAPConnection
15
db_conn: DBConnection
16

17

18
def get_user_pass(req: Tuple[Tuple[str, str]]) -> Tuple[str|None, str|None]:
19
    user_name = None
20
    pwd = None
21
    for item in req:
22
        if not item:
23
            continue
24
        for key, value in item:
25
            if key == "User-Name":
26
                user_name = value
27
            if key == "User-Password":
28
                pwd = value
29
    return (user_name, pwd)
30

31

32
# return 0 for success or -1 for failure
33
def instantiate(p):
34
    global cfg
35
    global ldap_conn
36
    global db_conn
37
    print("*** instantiate strongpass_otp ***")
38
    log_msg(radiusd.L_INFO, "*** instantiate strongpass_otp ***")
39
    print(p)
40
    try:
41
        with open(CONFIG_FILE, encoding='utf8') as file:
42
            cfg = yaml.load(file, Loader=yaml.FullLoader)
43
        ldap_conn = LDAPConnection(cfg["ldap"])
44
        db_conn = DBConnection(cfg["db"])
45
        return 0
46
    except Exception as e:
47
        log_msg(radiusd.L_ERR, e)
48
        return -1
49

50

51
def decode_ovpn_pass_otp(s: str)-> Tuple[bool, str | None, str| None]:
52
    parts: List[str] = s.split(':')
53
    if len(parts) != 3:
54
        log_msg(radiusd.L_ERR,
55
                       "wrong password format. Password format must be SCRV1:x:y. Where x,y base64 encoded values")
56
        return (False, None, None)
57
    if parts[0] != "SCRV1":
58
        log_msg(radiusd.L_ERR,
59
                       "wrong password format. Header 'SCRV1' is absent")
60
        return (False, None, None)
61
    passwd = b64decode(parts[1]).decode("utf-8")
62
    otp = b64decode(parts[2]).decode("utf-8")
63
    return (True, passwd, otp)
64

65

66
def decode_pass_and_otp(s: str)-> Tuple[bool, str | None, str | None]:
67
    if len(s)<7:
68
        log_msg(radiusd.L_ERR,
69
                       "Строка 'пароль плюс OTP код' слишком короткая. Минимальная длина должна быть 7 символов. 1 символ для пароля и 6 символов для OTP кода")
70
        return (False, None, None)
71
    
72
    passwd = s[:len(s)-6]
73
    otp=s[len(s)-6:]
74
    return (True, passwd, otp)
75

76

77
def decode_pwd(s: str) -> Tuple[bool, str | None, str| None]:
78
    global cfg
79
    match cfg['otp_option']:
80
        case 1:
81
            return decode_ovpn_pass_otp(s)
82
        case 2:
83
            return decode_pass_and_otp(s)
84
        case 3:
85
            return (True, "", s)
86
        case _:
87
            log_msg(radiusd.L_ERR,
88
                        "Параметр otp_option в конфигурационном файле содержит недопустимое значение")
89
            return (False, None, None)
90

91

92
def auth_via_ldap_and_otp(user:str, pwd: str, otp: str) -> Literal[0]|Literal[2]:
93
    try:
94
        otp_secret = db_conn.get_otp_secret(user)
95
        if otp_secret is None:
96
            log_msg(radiusd.L_ERR,
97
                           "OTP secret for user {} not found in database.".format(user))
98
            return radiusd.RLM_MODULE_REJECT
99
        totp = pyotp.TOTP(otp_secret)
100
        if not totp.verify(otp):
101
            log_msg(radiusd.L_ERR,
102
                           "OTP код для пользователя {} не прошел проверку. Отказ в аутентификации".format(user))
103
            return radiusd.RLM_MODULE_REJECT
104
        isAuthenticated = ldap_conn.authenticate(user, pwd)
105
        if isAuthenticated is False:
106
            log_msg(radiusd.L_ERR,
107
                           "user {} has not been authenticated. Wrong password or user not found".format(user))
108
            return radiusd.RLM_MODULE_REJECT
109
        return radiusd.RLM_MODULE_OK
110
    except Exception as ex:
111
        log_msg(radiusd.L_ERR,
112
                       "Unhandled exception when trying to authenticate user {}. Exception message: {}".format(user, ex))
113
        return radiusd.RLM_MODULE_REJECT
114

115
def auth_via_otp_only(user: str, otp: str) -> Literal[0]| Literal[2]:
116
    try:
117
        otp_secret = db_conn.get_otp_secret(user)
118
        if otp_secret is None:
119
            log_msg(radiusd.L_ERR,
120
                           "OTP secret for user {} not found in database.".format(user))
121
            return radiusd.RLM_MODULE_REJECT
122
        totp = pyotp.TOTP(otp_secret)
123
        if not totp.verify(otp):
124
            log_msg(radiusd.L_ERR,
125
                           "OTP код для пользователя {} не прошел проверку. Отказ в аутентификации".format(user))
126
            return radiusd.RLM_MODULE_REJECT
127
        return radiusd.RLM_MODULE_OK
128
    except Exception as ex:
129
        log_msg(radiusd.L_ERR,
130
                       "Unhandled exception when trying to authenticate user {}. Exception message: {}".format(user, ex))
131
        return radiusd.RLM_MODULE_REJECT
132

133
def auth_test_user(user: str, pwd: str)->Literal[0] | Literal[2]:
134
    if cfg['otp_option'] != 1 and cfg['otp_option'] != 2:
135
        log_msg(radiusd.L_ERR,
136
                    "Тестовый пользователь при этом значении otp_option не может быть аутентифицирован")
137
        return radiusd.RLM_MODULE_REJECT
138
    isAuthenticated = ldap_conn.authenticate(user, pwd)
139
    if isAuthenticated is False:
140
        log_msg(radiusd.L_ERR,
141
                        "user {} has not been authenticated. Wrong password or user not found".format(user))
142
        return radiusd.RLM_MODULE_REJECT
143
    return radiusd.RLM_MODULE_OK
144

145

146
def authenticate(p):
147
    global cfg
148
    try:
149
        user, p = get_user_pass(p)
150
        if user is None or p is None:
151
            return radiusd.RLM_MODULE_REJECT
152
        if user == cfg['auth_test_user']:
153
            return auth_test_user(user, p)
154
        result, passwd, otp = decode_pwd(p)
155
        if result is False:
156
            return radiusd.RLM_MODULE_REJECT
157
        match cfg['otp_option']:
158
            case 1:
159
                return auth_via_ldap_and_otp(user, passwd, otp)
160
            case 2:
161
                return auth_via_ldap_and_otp(user, passwd, otp)
162
            case 3:
163
                return auth_via_otp_only(user, otp)
164
            case _:
165
                log_msg(radiusd.L_ERR,
166
                            "Параметр otp_option в конфигурационном файле содержит недопустимое значение")
167
                return radiusd.RLM_MODULE_REJECT
168
    except Exception as ex:
169
        log_msg(radiusd.L_ERR,
170
                       "Unhandled exception when trying to authenticate user {}. Exception message: {}".format(user, ex))
171
        return radiusd.RLM_MODULE_REJECT
172

173

174
def authorize(p):
175
    user, pwd = get_user_pass(p)
176
    if user is None or pwd is None:
177
        return radiusd.RLM_MODULE_REJECT
178
    return (radiusd.RLM_MODULE_OK, (), (('Auth-Type', 'PAP'),))
179

180

181
def preacct(p):
182
    return radiusd.RLM_MODULE_OK
183

184

185
def accounting(p):
186
    return radiusd.RLM_MODULE_OK
187

188

189
def pre_proxy(p):
190
    return radiusd.RLM_MODULE_OK
191

192

193
def post_proxy(p):
194
    return radiusd.RLM_MODULE_OK
195

196

197
def post_auth(p):
198
    return radiusd.RLM_MODULE_OK
199

200

201
def recv_coa(p):
202
    return radiusd.RLM_MODULE_OK
203

204

205
def send_coa(p):
206
    return radiusd.RLM_MODULE_OK
207

208

209
def detach():
210
    return radiusd.RLM_MODULE_OK
211