demo-injection-sql-authentification-python/serveur.py at main · cedricvanrompay/demo-injection-sql-authentification-python · GitHub

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
import http.server
import json
import socketserver
import urllib.parse

import authentification
import base_de_donnees


PORT = 8000

CHEMIN_FICHIERS_STATIQUES = "fichiers-statiques"


base_de_donnees.initialisation()


class Handler(http.server.SimpleHTTPRequestHandler):
    def __init__(self, *args):
        super().__init__(*args, directory=CHEMIN_FICHIERS_STATIQUES)

    def do_POST(self):
        path = urllib.parse.urlparse(self.path)

        if path.path == "/authentification":
            content_length = int(self.headers.get('Content-Length', 0))
            set_cookie = authentification.traiter_requete_connection(self.rfile.read(content_length))

            if not set_cookie:
                self.send_error(http.HTTPStatus.UNAUTHORIZED)
                return

            self.send_response(http.HTTPStatus.FOUND)
            self.send_header("Set-Cookie", set_cookie)
            self.send_header("Location", "/messagerie.html")
            self.send_header("Content-Length", "0")
            self.end_headers()

        else:
            self.send_error(http.HTTPStatus.NOT_FOUND)

    def do_GET(self):
        path = urllib.parse.urlparse(self.path)
        query = urllib.parse.parse_qs(path.query)

        if path.path == "/api/messages":
            self.liste_messages(query)
        elif path.path == "/api/utilisateurs":
            self.liste_utilisateurs(query)
        else:
            super().do_GET()

    def liste_messages(self, query):
        utilisateur = authentification.authentifier_headers(self.headers["cookie"])

        if not utilisateur:
            self.send_error(http.HTTPStatus.UNAUTHORIZED)
            return

        if "correspondant" not in query:
            self.send_error(http.HTTPStatus.BAD_REQUEST)
            return

        correspondant = query["correspondant"][0]

        messages = base_de_donnees.lister_messages(
            auteur=utilisateur,
            destinataire=correspondant,
        )

        messages = messages + base_de_donnees.lister_messages(
            auteur=correspondant,
            destinataire=utilisateur,
        )

        messages.sort(key=lambda a: a["date"])


        self.send_response(http.HTTPStatus.OK)
        self.send_header("Content-Type", "application/json")
        self.end_headers()

        corps_reponse = json.dumps(messages, indent=4, ensure_ascii=False)

        self.wfile.write(corps_reponse.encode())

    def liste_utilisateurs(self, query):
        pseudo = query["pseudo"][0]

        utilisateurs = base_de_donnees.trouver_utilisateur(pseudo)

        self.send_response(http.HTTPStatus.OK)
        self.send_header("Content-Type", "application/json")
        self.end_headers()

        corps_reponse = json.dumps(utilisateurs, indent=4, ensure_ascii=False)

        self.wfile.write(corps_reponse.encode())


class ReusableTCPServer(socketserver.TCPServer):
    allow_reuse_address = True


with ReusableTCPServer(("", PORT), Handler) as httpd:
    print(f"service en cours à l'adresse http://localhost:{PORT}")
    httpd.serve_forever()