Ciemna strona mocy, backend, szyfrowanie żądań przy użyciu RSA w Python Flask - część II

Wprowadzenie

W poprzednich artykułach wyjaśniłem, dlaczego zdecydowałem się opisać poniższe implementacje. Dzisiaj chciałbym zakończyć serię postów opisujących mechanizmy szyfrowania i podpisywania żądań kluczami RSA. Dla zainteresowanych implementacją wspomnianych mechanizmów przygotowałem bibliotekę Flask-RSA, która jest również dostępna do pobrania z menedżera pakietów PyPI.

W kwestii przypomienia, stworzenie tej serii artykułów o szyfrowaniu i podpisywaniu żądań w API jest związane z projektem, który miałem przyjemność testować. Mechanizmy tu przedstawione są używane w rzeczywistym API do komunikacji między backendem a aplikacją mobilną. Autor projektu, oprócz mechanizmu podpisywania żądań (nagłówek X-Signature) i ochrony przed atakami replay (nagłówki X-Nonce-Value i X-Nonce-Created-At), które opisałem w wcześniejszych artykułach, zdecydował się również na użycie szyfrowania treści żądań przy użyciu kluczy RSA. Wszystkie te zabezpieczenia znacznie komplikują badanie komunikacji przez atakujących, szczególnie przy braku wewnętrznej wiedzy o implementacji.

Wróćmy do głównego tematu, szyfrowanie RSA jest asymetryczną metodą szyfrowania, która używa pary kluczy: publicznego i prywatnego. Klucz publiczny służy do szyfrowania danych, które następnie mogą być odszyfrowane tylko przy użyciu odpowiadającego im klucza prywatnego. W kontekście Flaska, proces szyfrowania żądań RSA rozpoczyna się od użycia publicznego klucza serwera do szyfrowania treści żądania wysyłanego przez klienta. W poniżej przedstawionej implementacji, serwer dodatkowo szyfruje odpowiedź wysyłaną do klienta, używając jego publicznego klucza. W praktyce oznacza to, że całą komunikację mogą zrozumieć tylko komunikujące się strony. Wymaga to jednak, aby obie strony wymieniły się swoimi kluczami publicznymi.

Implementacja szyfrowania w Flasku wymaga przede wszystkim wygenerowania pary kluczy RSA. Można to zrobić za pomocą narzędzi takich jak OpenSSL lub bezpośrednio w Pythonie, używając biblioteki pycryptodome. Po wygenerowaniu klucz publiczny musi zostać udostępniony nadawcy wiadomości, podczas gdy klucz prywatny jest bezpiecznie przechowywany po stronie odbiorcy.

Podczas implementacji szyfrowania RSA i podpisywania żądań, deweloperzy mogą napotkać kilka wyzwań, takich jak zarządzanie kluczami czy ochrona przed atakami replay. Ochronę przed atakami replay można osiągnąć, dodając do żądania wartość nonce i znacznik czasu, które są następnie weryfikowane po stronie serwera. Zapewnia to, że żądanie jest unikalne i zostało wysłane w odpowiednim oknie czasowym. Dodatkowo, aby zapewnić integralność wiadomości, dołączony zostanie podpis wykonany przy użyciu prywatnego klucza nadawcy wiadomości.

Opis zaimplementowanego algorytmu

W tej sekcji chciałbym przypomnieć czytelnikowi, jak mechanizm opisany w tym artykule różni się od tego, który został opisany w “Przykład implementacji walidacji podpisu żądania RSA w Python Flask”. Obecny proces zawiera dodatkowe kroki związane z szyfrowaniem i deszyfrowaniem treści żądania. Nowa wersja mechanizmu przedstawia się następująco:

1. Aplikacja używa publicznego klucza otrzymanego od serwera do szyfrowania treści żądania.
2. Aplikacja tworzy ustandaryzowany schemat żądania, który zawiera zaszyfrowaną wiadomość.
3. Aplikacja dodaje do żądania unikalną wartość nonce, generowaną za pomocą uuid4 (nagłówek X-Nonce-Value).
4. Aplikacja dołącza aktualną datę do żądania (nagłówek X-Nonce-Created-At).
5. Aplikacja tworzy sygnaturę żądania, używając swojego klucza prywatnego, treści wiadomości, ścieżki, wartości nonce i aktualnej daty.
6. Aplikacja dołącza wygenerowaną sygnaturę do żądania (nagłówek X-Signature).
7. Serwer API sprawdza poprawność sygnatury, używając publicznego klucza klienta.
8. Jeśli sygnatura jest ważna, serwer API akceptuje żądanie. W przypadku nieważnej sygnatury żądanie jest odrzucane.
9. Serwer API deszyfruje wiadomość, używając swojego klucza prywatnego.

W tym miejscu chciałbym przypomnieć, że analizowany mechanizm nie opisuje procedury dotyczącej wymiany kluczy publicznych między stronami komunikacji. Zakładam, że osoby wdrażające ten mechanizm rozważyły już wszystkie dostępne metody wymiany kluczy.

W następnej części artykułu skupimy się na uproszczeniu całego procesu przez użycie tylko jednej pary kluczy RSA.

Implementacja

Poniżej znajduje się implementacja dekoratora, którego celem jest deszyfrowanie treści przychodzących żądań, które zostały zaszyfrowane przy użyciu publicznego klucza serwera, i przekazanie odszyfrowanej treści do oryginalnej funkcji obsługującej żądanie. Przedstawiony proces deszyfrowania otrzymanego żądania wygląda następująco:

1. Odbiór żądania: funkcja oczekuje na żądanie wysłane w formacie JSON. Używając metody request.get_json(), pobiera treść żądania.
2. Sprawdzanie ważności wiadomości: Funkcja sprawdza, czy otrzymane żądanie jest w poprawnym formacie i czy ciało wiadomości JSON zawiera określony klucz (self._encrypted_payload_key), pod którym oczekiwana jest zaszyfrowana wiadomość.
3. Deszyfrowanie wiadomości: Jeśli odpowiedni klucz znajduje się w żądaniu, payload jest deszyfrowany przy użyciu prywatnego klucza serwera (_server_private_key). Do deszyfrowania używany jest algorytm OAEP z MGF1 i SHA-256 jako funkcją haszującą.
4. Przetwarzanie żądania: Odszyfrowana treść jest następnie dekodowana z formatu base64 i przekształcana z powrotem na obiekt JSON (request_body), który jest przekazywany do oryginalnej funkcji obsługującej żądanie jako argument.
5. Obsługa wyjątków: W przypadku błędów podczas procesu deszyfrowania, dekorator zwraca informacje o napotkanym problemie.
6. Niepoprawny format zaszyfrowanej wiadomości: Jeśli format otrzymanej wiadomości jest nieprawidłowy, dekorator zwraca informującą o napotkanym problemie.

Kod implementujący powyższe kroki przedstawia się następująco:

def decrypt_body(self):
    def _decrypt_body(f):
        @wraps(f)
        def decorator(*args, **kwargs):
            msg = request.get_json()
            if self._encrypted_payload_key in msg:

                try:
                    plaintext = _server_private_key.decrypt(
                        base64.standard_b64decode(msg[self._encrypted_payload_key]),
                        padding.OAEP(
                            mgf=padding.MGF1(algorithm=hashes.SHA256()),
                            algorithm=hashes.SHA256(),
                            label=None
                        )
                    )
                    request_body = json.loads(base64.standard_b64decode(plaintext))
                    return f(request_body, *args, **kwargs)
                except Exception as e:  # pylint: disable=broad-exception-caught
                    self._logger.error('Decryption problem %s', e)
                    return self._make_response('Decryption problem')
            else:
                return self._make_response(f'Missing {self._encrypted_payload_key} param')

        return decorator

    return _decrypt_body

Druga implementacja, którą chcę przedstawić, to dekorator zaprojektowany do szyfrowania odpowiedzi serwera przed wysłaniem jej do klienta. Głównym celem tego dekoratora jest zapewnienie, że wszystkie dane wysyłane z serwera do klienta są szyfrowane, co zwiększa bezpieczeństwo transmisji danych, szczególnie w kontekście informacji wrażliwych. Oto jak działa ten dekorator:

1. Wywoływanie funkcji odpowiedzi: Dekorator najpierw pozwala oryginalnej funkcji obsługującej żądanie na wykonanie swojej logiki i wygenerowanie odpowiedzi (response).
2. Szyfrowanie danych odpowiedzi: Następnie, używając publicznego klucza klienta (client_public_key), dekorator szyfruje dane zawarte w odpowiedzi. Szyfrowanie jest wykonywane przy użyciu algorytmu OAEP z MGF1 i SHA-256 jako funkcją haszującą. Dane odpowiedzi są najpierw kodowane do formatu base64, a następnie szyfrowane.
3. Przygotowywanie zaszyfrowanej wiadomości: Zaszyfrowany tekst (ciphertext) jest następnie dodawany do zdefiniowanej wcześniej struktury odpowiedzi (self._encrypted_payload_structure), pod określonym kluczem (self._encrypted_payload_key). Cała struktura jest kodowana do formatu base64 i umieszczana w odpowiedzi jako dane JSON.
4. Zwracanie zaszyfrowanej odpowiedzi: W końcu zmodyfikowane dane odpowiedzi są przypisywane z powrotem do obiektu odpowiedzi (response.data), a cała zaszyfrowana odpowiedź jest zwracana do klienta.

Kod implementujący powyższe kroki przedstawia się następująco:

def encrypt_body(self):
    def _encrypt_body(f):
        @wraps(f)
        def decorator(*args, **kwargs):
            response = f(*args, **kwargs)
            ciphertext = client_public_key.encrypt(
                base64.standard_b64encode(response.data),
                padding.OAEP(
                    mgf=padding.MGF1(algorithm=hashes.SHA256()),
                    algorithm=hashes.SHA256(),
                    label=None
                )
            )
            encrypted_msg = copy.deepcopy(self._encrypted_payload_structure)
            encrypted_msg[self._encrypted_payload_key] = base64.standard_b64encode(
                ciphertext).decode()
            response.data = json.dumps(encrypted_msg)
            return response
        return decorator
    return _encrypt_body

Przykład użycia

W tej sekcji przedstawię, jak działają powyższe implementacje. Zaczniemy od pierwszego dekoratora, @encrypted_request. Poniżej znajduje się kod naszego endpointa, który będziemy testować.

@app.route("/encrypted-body", methods=["POST"])
@decrypt_body
def encrypted_body():
    return jsonify({"msg": "Hello {}".format(request.data['name'])})

Aby przetestować funkcjonowanie prezentowanej implementacji, użyłem oprogramowania ZAP oraz skryptów, które opisałem w tych artykułach. Te skrypty są również dostępne w repozytorium community-scripts.

Rysunek 1 pokazuje wysyłanie żądania w jawny - niezaszyfrowany sposób, co zgodnie z implementacją jest przetwarzane przez serwer. Jak widać w odpowiedzi otrzymanej od serwera, przetwarzanie żądania nie powiodło się, serwer oczekuje wiadomości z polem encryptedPayload.

Rysunek 2 pokazuje to samo żądanie wysłane, tym razem przy użyciu napisanego skryptu. Jak wskazuje odpowiedź otrzymana od serwera, żądanie zostało prawidłowo przetworzone.

Rysunek 3 pokazuje rzeczywiste żądanie wysłane do serwera, które zostało zalogowane w historii ZAP. Możemy zauważyć, że treść wiadomości została całkowicie zmieniona na format oczekiwany przez nasz serwer.

Jak wspomniałem wcześniej, w następnym kroku, aby chronić się przed atakami typu replay, do wysłanej wiadomości zostanie dodany podpis przy użyciu klucza RSA. Poniżej znajduje się kod, który będziemy testować.

@app.route("/signed-encrypted-body", methods=["POST"])
@signature_required
@decrypt_body
def signed_encrypted_body():
    return jsonify({"msg": "Hello {}".format(request.data['name'])})

Rysunek 4 pokazuje wysłane żądanie, które nie zawiera wcześniej wspomnianych nagłówków odpowiedzialnych za nasz podpis, a mianowicie X-Nonce-Value, X-Nonce-Created-At i X-Signature.

Rysunek 5 pokazuje wysłane żądanie, które jest poprawnie interpretowane przez serwer.

Rysunek 6 pokazuje wysłane żądanie zarejestrowane w historii. Jak możemy zauważyć, wiadomość wysłana do serwera została zaszyfrowana. Dodatkowo, dodane zostały nagłówki X-Nonce-Value i X-Nonce-Created-At, a wiadomość została podpisana.

W ostatnim kroku dodamy szyfrowanie odpowiedzi z serwera, korzystając z ostatniego dekoratora przedstawionego w tym artykule, mianowicie @encrypt_body. Kod będzie wyglądał następująco:

@app.route('/encrypted-req-resp-signed', methods=['POST'])
@signature_required
@decrypt_body
@encrypt_body
def encrypted_endpoint():
    return jsonify({"msg": "Hello {}".format(request.data['name'])})

Rysunek 7 pokazuje poprawnie zaszyfrowaną i podpisaną wiadomość. Serwer ją przetworzył i zaszyfrował odpowiedź.

Podsumowanie

W tej serii artykułów skupiłem się na zabezpieczaniu usług backendowych za pomocą zaawansowanych technik szyfrowania oraz podpisywania żądań kluczami RSA w środowisku Flask. Głównym celem było nie tylko zapewnienie poufności danych poprzez szyfrowanie, ale także ich integralności i autentyczności przy użyciu sygnatur RSA. Prezentowane rozwiązania chronią komunikację przed różnymi zagrożeniami cybernetycznymi, w tym atakami typu replay, oferując dodatkową warstwę bezpieczeństwa standardowemu szyfrowaniu HTTPS.

Przeanalizowałem proces szyfrowania i deszyfrowania danych przesyłanych między klientem a serwerem, jak również integrację z podpisem, w celu ułatwienia identyfikacji i weryfikacji autentyczności danych. W tym kontekście przedstawiłem praktyczne implementacje dekoratorów we frameworku Flask, który automatyzuje procesy szyfrowania odpowiedzi serwera i deszyfrowania przychodzących żądań, zapewniając ochronę danych na każdym etapie komunikacji.

Istotnym elementem jest wymiana kluczy publicznych między klientem a serwerem, co pozwala na szyfrowanie i deszyfrowanie informacji w taki sposób, że dostęp do oryginalnej treści wiadomości mają tylko upoważnione strony. Użycie kluczy RSA umożliwia stosowanie silnych mechanizmów szyfrowania asymetrycznego, które są trudniejsze do złamania w porównaniu z metodami symetrycznymi.

Omówiłem również wyzwania związane z zarządzaniem kluczami i ochroną przed atakami typu replay, podkreślając znaczenie stosowania wartości nonce i znaczników czasu, aby zapewnić unikalność i aktualność żądań. Ponadto wskazałem na potrzebę używania podpisów, aby zapewnić integralność i autentyczność przesyłanych danych.

Bibliografia

1. https://en.wikipedia.org/wiki/Man-in-the-middle_attack
2. https://en.wikipedia.org/wiki/HMAC
3. https://en.wikipedia.org/wiki/Replay_attack
4. https://httpwg.org/http-extensions/draft-ietf-httpbis-message-signatures.1.html#name-rsassa-pkcs1-v1_5-using-sha
5. https://www.pycryptodome.org/
6. https://flask.palletsprojects.com/
7. https://circleci.com/blog/authentication-decorators-flask/
8. https://en.wikipedia.org/wiki/Don%27t_repeat_yourself
9. https://mwalkowski.com/pl/post/example-of-implementing-rsa-request-signature-validation-in-python-flask/
10. https://mwalkowski.com/pl/post/once-again-zap-scripting-request-encryption-and-finally-signing/
11. https://mwalkowski.com/pl/post/using-burp-python-scripts-to-encrypt-requests-with-rsa-keys/