forked from nginxinc/nginx-ldap-auth
-
Notifications
You must be signed in to change notification settings - Fork 0
/
backend-sample-app.py
executable file
·154 lines (115 loc) · 4.67 KB
/
backend-sample-app.py
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
#!/bin/sh
''''which python >/dev/null && exec python "$0" "$@" # '''
# Copyright (C) 2014-2015 Nginx, Inc.
# Example of an application working on port 9000
# To interact with nginx-ldap-auth-daemon this application
# 1) accepts GET requests on /login and responds with a login form
# 2) accepts POST requests on /login, sets a cookie, and responds with redirect
import sys, os, signal, base64, cgi
if sys.version_info.major == 2:
from urlparse import urlparse
from Cookie import BaseCookie
from BaseHTTPServer import HTTPServer, BaseHTTPRequestHandler
elif sys.version_info.major == 3:
from urllib.parse import urlparse
from http.cookies import BaseCookie
from http.server import HTTPServer, BaseHTTPRequestHandler
Listen = ('localhost', 9000)
import threading
if sys.version_info.major == 2:
from SocketServer import ThreadingMixIn
elif sys.version_info.major == 3:
from socketserver import ThreadingMixIn
def ensure_bytes(data):
return data if sys.version_info.major == 2 else data.encode("utf-8")
class AuthHTTPServer(ThreadingMixIn, HTTPServer):
pass
class AppHandler(BaseHTTPRequestHandler):
def do_GET(self):
url = urlparse(self.path)
if url.path.startswith("/login"):
return self.auth_form()
self.send_response(200)
self.end_headers()
self.wfile.write(ensure_bytes('Hello, world! Requested URL: ' + self.path + '\n'))
# send login form html
def auth_form(self, target = None):
# try to get target location from header
if target == None:
target = self.headers.get('X-Target')
# form cannot be generated if target is unknown
if target == None:
self.log_error('target url is not passed')
self.send_response(500)
return
html="""
<!DOCTYPE HTML PUBLIC "-//W3C//DTD HTML 4.01 Transitional//EN" "http://www.w3.org/TR/html4/loose.dtd">
<html>
<head>
<meta http-equiv=Content-Type content="text/html;charset=UTF-8">
<title>Auth form example</title>
</head>
<body>
<form action="/login" method="post">
<table>
<tr>
<td>Username: <input type="text" name="username"/></td>
<tr>
<td>Password: <input type="password" name="password"/></td>
<tr>
<td><input type="submit" value="Login"></td>
</table>
<input type="hidden" name="target" value="TARGET">
</form>
</body>
</html>"""
self.send_response(200)
self.end_headers()
self.wfile.write(ensure_bytes(html.replace('TARGET', target)))
# processes posted form and sets the cookie with login/password
def do_POST(self):
# prepare arguments for cgi module to read posted form
env = {'REQUEST_METHOD':'POST',
'CONTENT_TYPE': self.headers['Content-Type'],}
# read the form contents
form = cgi.FieldStorage(fp = self.rfile, headers = self.headers,
environ = env)
# extract required fields
user = form.getvalue('username')
passwd = form.getvalue('password')
target = form.getvalue('target')
if user != None and passwd != None and target != None:
# form is filled, set the cookie and redirect to target
# so that auth daemon will be able to use information from cookie
self.send_response(302)
# WARNING WARNING WARNING
#
# base64 is just an example method that allows to pack data into
# a cookie. You definitely want to perform some encryption here
# and share a key with auth daemon that extracts this information
#
# WARNING WARNING WARNING
enc = base64.b64encode(ensure_bytes(user + ':' + passwd))
if sys.version_info.major == 3:
enc = enc.decode()
self.send_header('Set-Cookie', b'nginxauth=' + enc + b'; httponly')
self.send_header('Location', target)
self.end_headers()
return
self.log_error('some form fields are not provided')
self.auth_form(target)
def log_message(self, format, *args):
if len(self.client_address) > 0:
addr = BaseHTTPRequestHandler.address_string(self)
else:
addr = "-"
sys.stdout.write("%s - - [%s] %s\n" % (addr,
self.log_date_time_string(), format % args))
def log_error(self, format, *args):
self.log_message(format, *args)
def exit_handler(signal, frame):
sys.exit(0)
if __name__ == '__main__':
server = AuthHTTPServer(Listen, AppHandler)
signal.signal(signal.SIGINT, exit_handler)
server.serve_forever()