diff --git a/src/Access/Authentication.cpp b/src/Access/Authentication.cpp
index 6bc9aeec4c2..7bbf8ec5efa 100644
--- a/src/Access/Authentication.cpp
+++ b/src/Access/Authentication.cpp
@@ -5,6 +5,7 @@
 #include <Access/LDAPClient.h>
 #include <Access/GSSAcceptor.h>
 #include <Common/Exception.h>
+#include <Poco/Net/SecureStreamSocketImpl.h>
 #include <Poco/SHA1Engine.h>
 #include <Common/typeid_cast.h>
 
@@ -14,6 +15,7 @@ namespace DB
 namespace ErrorCodes
 {
     extern const int NOT_IMPLEMENTED;
+    extern const int WRONG_PASSWORD;
 }
 
 namespace
@@ -67,6 +69,23 @@ namespace
     }
 }
 
+std::string getPeerCertificateCommonName(const Poco::Net::SocketImpl * socketImpl)
+{
+    std::string cn;
+
+    if (socketImpl->secure())
+    {
+        // expect socket is an instance of SecureStreamSocket
+        const Poco::Net::SecureStreamSocketImpl * secureSocketImpl = dynamic_cast<const Poco::Net::SecureStreamSocketImpl *>(socketImpl);
+        if (secureSocketImpl && secureSocketImpl->havePeerCertificate())
+        {
+            Poco::Crypto::X509Certificate cert = secureSocketImpl->peerCertificate();
+            cn = cert.commonName();
+        }
+    }
+
+    return cn;
+}
 
 bool Authentication::areCredentialsValid(const Credentials & credentials, const AuthenticationData & auth_data, const ExternalAuthenticators & external_authenticators)
 {
@@ -87,6 +106,9 @@ bool Authentication::areCredentialsValid(const Credentials & credentials, const
             case AuthenticationType::KERBEROS:
                 return external_authenticators.checkKerberosCredentials(auth_data.getKerberosRealm(), *gss_acceptor_context);
 
+            case AuthenticationType::SSL_CERTIFICATE:
+                throw Authentication::Require<BasicCredentials>("ClickHouse X.509 Authentication");
+
             case AuthenticationType::MAX:
                 break;
         }
@@ -110,6 +132,9 @@ bool Authentication::areCredentialsValid(const Credentials & credentials, const
             case AuthenticationType::KERBEROS:
                 throw Authentication::Require<BasicCredentials>("ClickHouse Basic Authentication");
 
+            case AuthenticationType::SSL_CERTIFICATE:
+                throw Authentication::Require<BasicCredentials>("ClickHouse X.509 Authentication");
+
             case AuthenticationType::MAX:
                 break;
         }
@@ -137,6 +162,35 @@ bool Authentication::areCredentialsValid(const Credentials & credentials, const
             case AuthenticationType::KERBEROS:
                 throw Authentication::Require<GSSAcceptorContext>(auth_data.getKerberosRealm());
 
+            case AuthenticationType::SSL_CERTIFICATE:
+                throw Authentication::Require<BasicCredentials>("ClickHouse X.509 Authentication");
+
+            case AuthenticationType::MAX:
+                break;
+        }
+    }
+
+    if (const auto * certificate_credentials = typeid_cast<const CertificateCredentials *>(&credentials))
+    {
+        switch (auth_data.getType())
+        {
+            case AuthenticationType::NO_PASSWORD:
+            case AuthenticationType::PLAINTEXT_PASSWORD:
+            case AuthenticationType::SHA256_PASSWORD:
+            case AuthenticationType::DOUBLE_SHA1_PASSWORD:
+            case AuthenticationType::LDAP:
+                throw Authentication::Require<BasicCredentials>("ClickHouse Basic Authentication");
+
+            case AuthenticationType::KERBEROS:
+                throw Authentication::Require<GSSAcceptorContext>(auth_data.getKerberosRealm());
+
+            case AuthenticationType::SSL_CERTIFICATE:
+                // N.B. the certificate should only be trusted when 'strict' SSL mode is enabled
+                if (!auth_data.containsSSLCertificateCommonName(certificate_credentials->getX509CommonName()))
+                    throw Exception("X.509 certificate is not on allowed list", ErrorCodes::WRONG_PASSWORD);
+
+                return true;
+
             case AuthenticationType::MAX:
                 break;
         }
diff --git a/src/Access/Authentication.h b/src/Access/Authentication.h
index 000ba8ca324..e081f8211f6 100644
--- a/src/Access/Authentication.h
+++ b/src/Access/Authentication.h
@@ -2,6 +2,7 @@
 
 #include <Access/Common/AuthenticationData.h>
 #include <Common/Exception.h>
+#include <Poco/Net/SocketImpl.h>
 #include <base/types.h>
 
 
@@ -15,6 +16,7 @@ namespace ErrorCodes
 class Credentials;
 class ExternalAuthenticators;
 
+std::string getPeerCertificateCommonName(const Poco::Net::SocketImpl * socketImpl);
 
 /// TODO: Try to move this checking to Credentials.
 struct Authentication
diff --git a/src/Access/Common/AuthenticationData.cpp b/src/Access/Common/AuthenticationData.cpp
index 7412d7336e3..6fee41c3dd3 100644
--- a/src/Access/Common/AuthenticationData.cpp
+++ b/src/Access/Common/AuthenticationData.cpp
@@ -59,6 +59,11 @@ const AuthenticationTypeInfo & AuthenticationTypeInfo::get(AuthenticationType ty
             static const auto info = make_info("KERBEROS");
             return info;
         }
+        case AuthenticationType::SSL_CERTIFICATE:
+        {
+            static const auto info = make_info("SSL_CERTIFICATE");
+            return info;
+        }
         case AuthenticationType::MAX:
             break;
     }
@@ -112,6 +117,7 @@ void AuthenticationData::setPassword(const String & password_)
         case AuthenticationType::NO_PASSWORD:
         case AuthenticationType::LDAP:
         case AuthenticationType::KERBEROS:
+        case AuthenticationType::SSL_CERTIFICATE:
             throw Exception("Cannot specify password for authentication type " + toString(type), ErrorCodes::LOGICAL_ERROR);
 
         case AuthenticationType::MAX:
@@ -149,7 +155,7 @@ void AuthenticationData::setPasswordHashHex(const String & hash)
 
 String AuthenticationData::getPasswordHashHex() const
 {
-    if (type == AuthenticationType::LDAP || type == AuthenticationType::KERBEROS)
+    if (type == AuthenticationType::LDAP || type == AuthenticationType::KERBEROS || type == AuthenticationType::SSL_CERTIFICATE)
         throw Exception("Cannot get password hex hash for authentication type " + toString(type), ErrorCodes::LOGICAL_ERROR);
 
     String hex;
@@ -194,6 +200,7 @@ void AuthenticationData::setPasswordHashBinary(const Digest & hash)
         case AuthenticationType::NO_PASSWORD:
         case AuthenticationType::LDAP:
         case AuthenticationType::KERBEROS:
+        case AuthenticationType::SSL_CERTIFICATE:
             throw Exception("Cannot specify password binary hash for authentication type " + toString(type), ErrorCodes::LOGICAL_ERROR);
 
         case AuthenticationType::MAX:
@@ -202,4 +209,18 @@ void AuthenticationData::setPasswordHashBinary(const Digest & hash)
     throw Exception("setPasswordHashBinary(): authentication type " + toString(type) + " not supported", ErrorCodes::NOT_IMPLEMENTED);
 }
 
+void AuthenticationData::clearAllowedCertificates()
+{
+    allowed_certificates.clear();
+}
+
+void AuthenticationData::addSSLCertificateCommonName(const String & x509CommonName)
+{
+    allowed_certificates.insert(x509CommonName);
+}
+
+bool AuthenticationData::containsSSLCertificateCommonName(const String & x509CommonName) const
+{
+    return allowed_certificates.find(x509CommonName) != allowed_certificates.end();
+}
 }
diff --git a/src/Access/Common/AuthenticationData.h b/src/Access/Common/AuthenticationData.h
index 8b390fd0900..152ba55b73d 100644
--- a/src/Access/Common/AuthenticationData.h
+++ b/src/Access/Common/AuthenticationData.h
@@ -1,6 +1,7 @@
 #pragma once
 
 #include <base/types.h>
+#include <set>
 #include <vector>
 
 namespace DB
@@ -27,6 +28,10 @@ enum class AuthenticationType
     /// Kerberos authentication performed through GSS-API negotiation loop.
     KERBEROS,
 
+    /// Authentication is done in SSL by checking user certificate.
+    /// Certificates may only be trusted if 'strict' SSL mode is enabled.
+    SSL_CERTIFICATE,
+
     MAX,
 };
 
@@ -79,6 +84,10 @@ public:
     const String & getKerberosRealm() const { return kerberos_realm; }
     void setKerberosRealm(const String & realm) { kerberos_realm = realm; }
 
+    void clearAllowedCertificates();
+    void addSSLCertificateCommonName(const String & x509CommonName);
+    bool containsSSLCertificateCommonName(const String & x509CommonName) const;
+
     friend bool operator ==(const AuthenticationData & lhs, const AuthenticationData & rhs);
     friend bool operator !=(const AuthenticationData & lhs, const AuthenticationData & rhs) { return !(lhs == rhs); }
 
@@ -97,6 +106,7 @@ private:
     Digest password_hash;
     String ldap_server_name;
     String kerberos_realm;
+    std::set<String> allowed_certificates;
 };
 
 }
diff --git a/src/Access/Credentials.cpp b/src/Access/Credentials.cpp
index c2850ad4d4f..79876cddcfe 100644
--- a/src/Access/Credentials.cpp
+++ b/src/Access/Credentials.cpp
@@ -48,6 +48,20 @@ void AlwaysAllowCredentials::setUserName(const String & user_name_)
     user_name = user_name_;
 }
 
+CertificateCredentials::CertificateCredentials(const String & user_name_, const String & x509CommonName_)
+    : Credentials(user_name_)
+    , x509CommonName(x509CommonName_)
+{
+    is_ready = true;
+}
+
+const String & CertificateCredentials::getX509CommonName() const
+{
+    if (!isReady())
+        throwNotReady();
+    return x509CommonName;
+}
+
 BasicCredentials::BasicCredentials()
 {
     is_ready = true;
diff --git a/src/Access/Credentials.h b/src/Access/Credentials.h
index 4992199f64c..f667d60bde3 100644
--- a/src/Access/Credentials.h
+++ b/src/Access/Credentials.h
@@ -38,6 +38,18 @@ public:
     void setUserName(const String & user_name_);
 };
 
+class CertificateCredentials
+    : public Credentials
+{
+public:
+    explicit CertificateCredentials(const String & user_name_, const String & x509CommonName_);
+
+    const String & getX509CommonName() const;
+
+private:
+    String x509CommonName;
+};
+
 class BasicCredentials
     : public Credentials
 {
diff --git a/src/Access/UsersConfigAccessStorage.cpp b/src/Access/UsersConfigAccessStorage.cpp
index 5bd2da97445..5d7933190d5 100644
--- a/src/Access/UsersConfigAccessStorage.cpp
+++ b/src/Access/UsersConfigAccessStorage.cpp
@@ -62,13 +62,16 @@ namespace
         bool has_ldap = config.has(user_config + ".ldap");
         bool has_kerberos = config.has(user_config + ".kerberos");
 
-        size_t num_password_fields = has_no_password + has_password_plaintext + has_password_sha256_hex + has_password_double_sha1_hex + has_ldap + has_kerberos;
+        const auto certificates_config = user_config + ".certificates";
+        bool has_certificates = config.has(certificates_config);
+
+        size_t num_password_fields = has_no_password + has_password_plaintext + has_password_sha256_hex + has_password_double_sha1_hex + has_ldap + has_kerberos + has_certificates;
         if (num_password_fields > 1)
-            throw Exception("More than one field of 'password', 'password_sha256_hex', 'password_double_sha1_hex', 'no_password', 'ldap', 'kerberos' are used to specify password for user " + user_name + ". Must be only one of them.",
+            throw Exception("More than one field of 'password', 'password_sha256_hex', 'password_double_sha1_hex', 'no_password', 'ldap', 'kerberos', 'certificates' are used to specify authentication info for user " + user_name + ". Must be only one of them.",
                 ErrorCodes::BAD_ARGUMENTS);
 
         if (num_password_fields < 1)
-            throw Exception("Either 'password' or 'password_sha256_hex' or 'password_double_sha1_hex' or 'no_password' or 'ldap' or 'kerberos' must be specified for user " + user_name + ".", ErrorCodes::BAD_ARGUMENTS);
+            throw Exception("Either 'password' or 'password_sha256_hex' or 'password_double_sha1_hex' or 'no_password' or 'ldap' or 'kerberos' or 'certificates' must be specified for user " + user_name + ".", ErrorCodes::BAD_ARGUMENTS);
 
         if (has_password_plaintext)
         {
@@ -105,6 +108,24 @@ namespace
             user->auth_data = AuthenticationData{AuthenticationType::KERBEROS};
             user->auth_data.setKerberosRealm(realm);
         }
+        else if (has_certificates)
+        {
+            user->auth_data = AuthenticationData{AuthenticationType::SSL_CERTIFICATE};
+
+            /// Fill list of allowed certificates.
+            Poco::Util::AbstractConfiguration::Keys keys;
+            config.keys(certificates_config, keys);
+            user->auth_data.clearAllowedCertificates();
+            for (const String & key : keys)
+            {
+                if (key.starts_with("common_name"))
+                {
+                    String value = config.getString(certificates_config + "." + key);
+                    user->auth_data.addSSLCertificateCommonName(value);
+                }
+                else
+                    throw Exception("Unknown certificate pattern type: " + key, ErrorCodes::BAD_ARGUMENTS);            }
+        }
 
         const auto profile_name_config = user_config + ".profile";
         if (config.has(profile_name_config))
diff --git a/src/Dictionaries/ClickHouseDictionarySource.cpp b/src/Dictionaries/ClickHouseDictionarySource.cpp
index bd9a1f7776e..deecc3c983e 100644
--- a/src/Dictionaries/ClickHouseDictionarySource.cpp
+++ b/src/Dictionaries/ClickHouseDictionarySource.cpp
@@ -277,7 +277,7 @@ void registerDictionarySourceClickHouse(DictionarySourceFactory & factory)
         {
             /// We should set user info even for the case when the dictionary is loaded in-process (without TCP communication).
             Session session(global_context, ClientInfo::Interface::LOCAL);
-            session.authenticate(configuration.user, configuration.password, {});
+            session.authenticate(configuration.user, configuration.password, Poco::Net::SocketAddress{});
             context = session.makeQueryContext();
         }
         else
diff --git a/src/Interpreters/Session.cpp b/src/Interpreters/Session.cpp
index 2af9a2b6bbc..d05f438bb80 100644
--- a/src/Interpreters/Session.cpp
+++ b/src/Interpreters/Session.cpp
@@ -1,6 +1,7 @@
 #include <Interpreters/Session.h>
 
 #include <Access/AccessControl.h>
+#include <Access/Authentication.h>
 #include <Access/Credentials.h>
 #include <Access/ContextAccess.h>
 #include <Access/User.h>
@@ -297,6 +298,30 @@ void Session::authenticate(const String & user_name, const String & password, co
     authenticate(BasicCredentials{user_name, password}, address);
 }
 
+void Session::authenticate(const String & user_name, const String & password, const Poco::Net::StreamSocket & socket)
+{
+    switch (getAuthenticationType(user_name))
+    {
+        case AuthenticationType::NO_PASSWORD:
+        case AuthenticationType::PLAINTEXT_PASSWORD:
+        case AuthenticationType::DOUBLE_SHA1_PASSWORD:
+        case AuthenticationType::SHA256_PASSWORD:
+        case AuthenticationType::LDAP:
+            authenticate(BasicCredentials{user_name, password}, socket.peerAddress());
+            break;
+
+        case AuthenticationType::KERBEROS:
+            throw Authentication::Require<BasicCredentials>("ClickHouse Basic Authentication");
+
+        case AuthenticationType::SSL_CERTIFICATE:
+            authenticate(CertificateCredentials{user_name, getPeerCertificateCommonName(socket.impl())}, socket.peerAddress());
+            break;
+
+        case AuthenticationType::MAX:
+            break;
+    }
+}
+
 void Session::authenticate(const Credentials & credentials_, const Poco::Net::SocketAddress & address_)
 {
     if (session_context)
diff --git a/src/Interpreters/Session.h b/src/Interpreters/Session.h
index 71964130412..b65de2f3533 100644
--- a/src/Interpreters/Session.h
+++ b/src/Interpreters/Session.h
@@ -4,6 +4,7 @@
 #include <Access/Common/AuthenticationData.h>
 #include <Interpreters/ClientInfo.h>
 #include <Interpreters/Context_fwd.h>
+#include <Poco/Net/StreamSocket.h>
 
 #include <chrono>
 #include <memory>
@@ -49,6 +50,7 @@ public:
     /// Sets the current user, checks the credentials and that the specified address is allowed to connect from.
     /// The function throws an exception if there is no such user or password is wrong.
     void authenticate(const String & user_name, const String & password, const Poco::Net::SocketAddress & address);
+    void authenticate(const String & user_name, const String & password, const Poco::Net::StreamSocket & socket);
     void authenticate(const Credentials & credentials_, const Poco::Net::SocketAddress & address_);
 
     /// Returns a reference to session ClientInfo.
diff --git a/src/Parsers/Access/ASTCreateUserQuery.cpp b/src/Parsers/Access/ASTCreateUserQuery.cpp
index 18030a5ed80..70f29a02d85 100644
--- a/src/Parsers/Access/ASTCreateUserQuery.cpp
+++ b/src/Parsers/Access/ASTCreateUserQuery.cpp
@@ -78,6 +78,7 @@ namespace
                 }
 
                 case AuthenticationType::NO_PASSWORD: [[fallthrough]];
+                case AuthenticationType::SSL_CERTIFICATE: [[fallthrough]];
                 case AuthenticationType::MAX:
                     throw Exception("AST: Unexpected authentication type " + toString(auth_type), ErrorCodes::LOGICAL_ERROR);
             }
diff --git a/src/Server/HTTP/HTTPServerRequest.cpp b/src/Server/HTTP/HTTPServerRequest.cpp
index f9ef14765c9..db0ccecf46f 100644
--- a/src/Server/HTTP/HTTPServerRequest.cpp
+++ b/src/Server/HTTP/HTTPServerRequest.cpp
@@ -9,9 +9,11 @@
 #include <Server/HTTP/HTTPServerResponse.h>
 #include <Server/HTTP/ReadHeaders.h>
 
+#include <Poco/Crypto/X509Certificate.h>
 #include <Poco/Net/HTTPHeaderStream.h>
 #include <Poco/Net/HTTPStream.h>
 #include <Poco/Net/NetException.h>
+#include <Poco/Net/SecureStreamSocketImpl.h>
 
 namespace DB
 {
@@ -69,6 +71,11 @@ bool HTTPServerRequest::checkPeerConnected() const
     return true;
 }
 
+Poco::Net::SocketImpl * HTTPServerRequest::getSocket() const
+{
+    return socket;
+}
+
 void HTTPServerRequest::readRequest(ReadBuffer & in)
 {
     char ch;
diff --git a/src/Server/HTTP/HTTPServerRequest.h b/src/Server/HTTP/HTTPServerRequest.h
index 33463177743..85fe3a6f80a 100644
--- a/src/Server/HTTP/HTTPServerRequest.h
+++ b/src/Server/HTTP/HTTPServerRequest.h
@@ -38,6 +38,8 @@ public:
     /// Returns the server's address.
     const Poco::Net::SocketAddress & serverAddress() const { return server_address; }
 
+    Poco::Net::SocketImpl * getSocket() const;
+
 private:
     /// Limits for basic sanity checks when reading a header
     enum Limits
diff --git a/src/Server/HTTPHandler.cpp b/src/Server/HTTPHandler.cpp
index 673edfb6719..118a929323d 100644
--- a/src/Server/HTTPHandler.cpp
+++ b/src/Server/HTTPHandler.cpp
@@ -322,10 +322,35 @@ bool HTTPHandler::authenticateUser(
     std::string user = request.get("X-ClickHouse-User", "");
     std::string password = request.get("X-ClickHouse-Key", "");
     std::string quota_key = request.get("X-ClickHouse-Quota", "");
+    std::string x509_auth = request.get("X-ClickHouse-X509Authentication", "");
 
     std::string spnego_challenge;
+    std::string certificate_common_name;
 
-    if (user.empty() && password.empty() && quota_key.empty())
+    LOG_DEBUG(log, "X-ClickHouse-X509Authentication=\"{}\"", x509_auth);
+
+    bool has_basic_auth = request.hasCredentials() || params.has("user") || params.has("password") || params.has("quota_key");
+    bool has_header_auth = !(user.empty() && password.empty() && quota_key.empty());
+    bool has_x509_auth = !user.empty() && (x509_auth == "yes"); // header values are case sensitive
+
+    if (has_x509_auth)
+    {
+        if (has_header_auth || has_basic_auth)
+            throw Exception("Invalid authentication: it is not allowed to use SSL X.509 certificate authentication and other authentication methods simultaneously", ErrorCodes::AUTHENTICATION_FAILED);
+
+        certificate_common_name = getPeerCertificateCommonName(request.getSocket());
+        if (certificate_common_name.empty())
+            throw Exception("Invalid authentication: empty X.509 certificate Common Name", ErrorCodes::AUTHENTICATION_FAILED);
+
+        LOG_DEBUG(log, "certificate_common_name=\"{}\"", certificate_common_name);
+    }
+    else if (has_header_auth)
+    {
+        /// It is prohibited to mix different authorization schemes.
+        if (has_basic_auth)
+            throw Exception("Invalid authentication: it is not allowed to use X-ClickHouse HTTP headers and other authentication methods simultaneously", ErrorCodes::AUTHENTICATION_FAILED);
+    }
+    else
     {
         /// User name and password can be passed using query parameters
         /// or using HTTP Basic auth (both methods are insecure).
@@ -365,26 +390,17 @@ bool HTTPHandler::authenticateUser(
 
         quota_key = params.get("quota_key", "");
     }
-    else
-    {
-        /// It is prohibited to mix different authorization schemes.
-        if (request.hasCredentials() || params.has("user") || params.has("password") || params.has("quota_key"))
-            throw Exception("Invalid authentication: it is not allowed to use X-ClickHouse HTTP headers and other authentication methods simultaneously", ErrorCodes::AUTHENTICATION_FAILED);
-    }
 
-    if (spnego_challenge.empty()) // I.e., now using user name and password strings ("Basic").
+    if (!certificate_common_name.empty())
     {
         if (!request_credentials)
-            request_credentials = std::make_unique<BasicCredentials>();
+            request_credentials = std::make_unique<CertificateCredentials>(user, certificate_common_name);
 
-        auto * basic_credentials = dynamic_cast<BasicCredentials *>(request_credentials.get());
-        if (!basic_credentials)
-            throw Exception("Invalid authentication: unexpected 'Basic' HTTP Authorization scheme", ErrorCodes::AUTHENTICATION_FAILED);
-
-        basic_credentials->setUserName(user);
-        basic_credentials->setPassword(password);
+        auto * certificate_credentials = dynamic_cast<CertificateCredentials *>(request_credentials.get());
+        if (!certificate_credentials)
+            throw Exception("Invalid authentication: expected SSL certificate authorization scheme", ErrorCodes::AUTHENTICATION_FAILED);
     }
-    else
+    else if (!spnego_challenge.empty())
     {
         if (!request_credentials)
             request_credentials = server.context()->makeGSSAcceptorContext();
@@ -411,6 +427,18 @@ bool HTTPHandler::authenticateUser(
             return false;
         }
     }
+    else // I.e., now using user name and password strings ("Basic").
+    {
+        if (!request_credentials)
+            request_credentials = std::make_unique<BasicCredentials>();
+
+        auto * basic_credentials = dynamic_cast<BasicCredentials *>(request_credentials.get());
+        if (!basic_credentials)
+            throw Exception("Invalid authentication: expected 'Basic' HTTP Authorization scheme", ErrorCodes::AUTHENTICATION_FAILED);
+
+        basic_credentials->setUserName(user);
+        basic_credentials->setPassword(password);
+    }
 
     /// Set client info. It will be used for quota accounting parameters in 'setUser' method.
     ClientInfo & client_info = session->getClientInfo();
diff --git a/src/Server/TCPHandler.cpp b/src/Server/TCPHandler.cpp
index 6fa2b25d181..d4112dd8bc7 100644
--- a/src/Server/TCPHandler.cpp
+++ b/src/Server/TCPHandler.cpp
@@ -1177,7 +1177,7 @@ void TCPHandler::receiveHello()
         return;
     }
 
-    session->authenticate(user, password, socket().peerAddress());
+    session->authenticate(user, password, socket());
 }