diff --git a/CHANGELOG.rst b/CHANGELOG.rst
index 3caf14d7bc..aa4439df03 100644
--- a/CHANGELOG.rst
+++ b/CHANGELOG.rst
@@ -139,6 +139,19 @@ Added
 
   Contributed by @Kami.
 
+* Add new ``credentials.basic_auth = username:password`` CLI configuration option.
+
+  This argument allows client to use additional set of basic auth credentials when talking to the
+  StackStorm API endpoints (api, auth, stream) - that is, in addition to the token / api key
+  native StackStorm auth.
+
+  This allows for simple basic auth based multi factor authentication implementation for
+  installations which don't utilize SSO.
+
+  #5152
+
+  Contributed by @Kami.
+
 Fixed
 ~~~~~
 
diff --git a/conf/st2rc.sample.ini b/conf/st2rc.sample.ini
index f431aa1f3a..5d62b59ef5 100644
--- a/conf/st2rc.sample.ini
+++ b/conf/st2rc.sample.ini
@@ -21,6 +21,11 @@ username = test1
 password = testpassword
 # or authenticate with an api key.
 # api_key = <key>
+# Optional additional http basic auth credentials which are sent with each HTTP
+# request except the auth request to /v1/auth/tokens endpoint.
+# Available in StackStorm >= v3.4.0
+# NOTE: Username can't contain colon (:) character.
+# basic_auth = username:password
 
 [api]
 url = http://127.0.0.1:9101/v1
diff --git a/st2client/st2client/base.py b/st2client/st2client/base.py
index 54b7a91b14..3c797ea20d 100644
--- a/st2client/st2client/base.py
+++ b/st2client/st2client/base.py
@@ -57,6 +57,7 @@
     "api_key": ["credentials", "api_key"],
     "cacert": ["general", "cacert"],
     "debug": ["cli", "debug"],
+    "basic_auth": ["credentials", "basic_auth"],
 }
 
 
@@ -87,6 +88,7 @@ def get_client(self, args, debug=False):
             "stream_url",
             "api_version",
             "cacert",
+            "basic_auth",
         ]
         cli_options = {opt: getattr(args, opt, None) for opt in cli_options}
         if cli_options.get("cacert", None) is not None:
diff --git a/st2client/st2client/client.py b/st2client/st2client/client.py
index 9772c825b7..ec2878758d 100644
--- a/st2client/st2client/client.py
+++ b/st2client/st2client/client.py
@@ -63,6 +63,7 @@ def __init__(
         debug=False,
         token=None,
         api_key=None,
+        basic_auth=None,
     ):
         # Get CLI options. If not given, then try to get it from the environment.
         self.endpoints = dict()
@@ -129,115 +130,195 @@ def __init__(
 
         self.api_key = api_key
 
+        if basic_auth:
+            # NOTE: We assume username can't contain colons
+            if len(basic_auth.split(":", 1)) != 2:
+                raise ValueError(
+                    "basic_auth config options needs to be in the "
+                    "username:password notation"
+                )
+
+            self.basic_auth = tuple(basic_auth.split(":", 1))
+        else:
+            self.basic_auth = None
+
         # Instantiate resource managers and assign appropriate API endpoint.
         self.managers = dict()
         self.managers["Token"] = ResourceManager(
-            models.Token, self.endpoints["auth"], cacert=self.cacert, debug=self.debug
+            models.Token,
+            self.endpoints["auth"],
+            cacert=self.cacert,
+            debug=self.debug,
+            basic_auth=self.basic_auth,
         )
         self.managers["RunnerType"] = ResourceManager(
             models.RunnerType,
             self.endpoints["api"],
             cacert=self.cacert,
             debug=self.debug,
+            basic_auth=self.basic_auth,
         )
         self.managers["Action"] = ActionResourceManager(
-            models.Action, self.endpoints["api"], cacert=self.cacert, debug=self.debug
+            models.Action,
+            self.endpoints["api"],
+            cacert=self.cacert,
+            debug=self.debug,
+            basic_auth=self.basic_auth,
         )
         self.managers["ActionAlias"] = ActionAliasResourceManager(
             models.ActionAlias,
             self.endpoints["api"],
             cacert=self.cacert,
             debug=self.debug,
+            basic_auth=self.basic_auth,
         )
         self.managers["ActionAliasExecution"] = ActionAliasExecutionManager(
             models.ActionAliasExecution,
             self.endpoints["api"],
             cacert=self.cacert,
             debug=self.debug,
+            basic_auth=self.basic_auth,
         )
         self.managers["ApiKey"] = ResourceManager(
-            models.ApiKey, self.endpoints["api"], cacert=self.cacert, debug=self.debug
+            models.ApiKey,
+            self.endpoints["api"],
+            cacert=self.cacert,
+            debug=self.debug,
+            basic_auth=self.basic_auth,
         )
         self.managers["Config"] = ConfigManager(
-            models.Config, self.endpoints["api"], cacert=self.cacert, debug=self.debug
+            models.Config,
+            self.endpoints["api"],
+            cacert=self.cacert,
+            debug=self.debug,
+            basic_auth=self.basic_auth,
         )
         self.managers["ConfigSchema"] = ResourceManager(
             models.ConfigSchema,
             self.endpoints["api"],
             cacert=self.cacert,
             debug=self.debug,
+            basic_auth=self.basic_auth,
         )
         self.managers["Execution"] = ExecutionResourceManager(
             models.Execution,
             self.endpoints["api"],
             cacert=self.cacert,
             debug=self.debug,
+            basic_auth=self.basic_auth,
         )
         # NOTE: LiveAction has been deprecated in favor of Execution. It will be left here for
         # backward compatibility reasons until v3.2.0
         self.managers["LiveAction"] = self.managers["Execution"]
         self.managers["Inquiry"] = InquiryResourceManager(
-            models.Inquiry, self.endpoints["api"], cacert=self.cacert, debug=self.debug
+            models.Inquiry,
+            self.endpoints["api"],
+            cacert=self.cacert,
+            debug=self.debug,
+            basic_auth=self.basic_auth,
         )
         self.managers["Pack"] = PackResourceManager(
-            models.Pack, self.endpoints["api"], cacert=self.cacert, debug=self.debug
+            models.Pack,
+            self.endpoints["api"],
+            cacert=self.cacert,
+            debug=self.debug,
+            basic_auth=self.basic_auth,
         )
         self.managers["Policy"] = ResourceManager(
-            models.Policy, self.endpoints["api"], cacert=self.cacert, debug=self.debug
+            models.Policy,
+            self.endpoints["api"],
+            cacert=self.cacert,
+            debug=self.debug,
+            basic_auth=self.basic_auth,
         )
         self.managers["PolicyType"] = ResourceManager(
             models.PolicyType,
             self.endpoints["api"],
             cacert=self.cacert,
             debug=self.debug,
+            basic_auth=self.basic_auth,
         )
         self.managers["Rule"] = ResourceManager(
-            models.Rule, self.endpoints["api"], cacert=self.cacert, debug=self.debug
+            models.Rule,
+            self.endpoints["api"],
+            cacert=self.cacert,
+            debug=self.debug,
+            basic_auth=self.basic_auth,
         )
         self.managers["Sensor"] = ResourceManager(
-            models.Sensor, self.endpoints["api"], cacert=self.cacert, debug=self.debug
+            models.Sensor,
+            self.endpoints["api"],
+            cacert=self.cacert,
+            debug=self.debug,
+            basic_auth=self.basic_auth,
         )
         self.managers["TriggerType"] = ResourceManager(
             models.TriggerType,
             self.endpoints["api"],
             cacert=self.cacert,
             debug=self.debug,
+            basic_auth=self.basic_auth,
         )
         self.managers["Trigger"] = ResourceManager(
-            models.Trigger, self.endpoints["api"], cacert=self.cacert, debug=self.debug
+            models.Trigger,
+            self.endpoints["api"],
+            cacert=self.cacert,
+            debug=self.debug,
+            basic_auth=self.basic_auth,
         )
         self.managers["TriggerInstance"] = TriggerInstanceResourceManager(
             models.TriggerInstance,
             self.endpoints["api"],
             cacert=self.cacert,
             debug=self.debug,
+            basic_auth=self.basic_auth,
         )
         self.managers["KeyValuePair"] = ResourceManager(
             models.KeyValuePair,
             self.endpoints["api"],
             cacert=self.cacert,
             debug=self.debug,
+            basic_auth=self.basic_auth,
         )
         self.managers["Webhook"] = WebhookManager(
-            models.Webhook, self.endpoints["api"], cacert=self.cacert, debug=self.debug
+            models.Webhook,
+            self.endpoints["api"],
+            cacert=self.cacert,
+            debug=self.debug,
+            basic_auth=self.basic_auth,
         )
         self.managers["Timer"] = ResourceManager(
-            models.Timer, self.endpoints["api"], cacert=self.cacert, debug=self.debug
+            models.Timer,
+            self.endpoints["api"],
+            cacert=self.cacert,
+            debug=self.debug,
+            basic_auth=self.basic_auth,
         )
         self.managers["Trace"] = ResourceManager(
-            models.Trace, self.endpoints["api"], cacert=self.cacert, debug=self.debug
+            models.Trace,
+            self.endpoints["api"],
+            cacert=self.cacert,
+            debug=self.debug,
+            basic_auth=self.basic_auth,
         )
         self.managers["RuleEnforcement"] = ResourceManager(
             models.RuleEnforcement,
             self.endpoints["api"],
             cacert=self.cacert,
             debug=self.debug,
+            basic_auth=self.basic_auth,
         )
         self.managers["Stream"] = StreamManager(
-            self.endpoints["stream"], cacert=self.cacert, debug=self.debug
+            self.endpoints["stream"],
+            cacert=self.cacert,
+            debug=self.debug,
+            basic_auth=self.basic_auth,
         )
         self.managers["Workflow"] = WorkflowManager(
-            self.endpoints["api"], cacert=self.cacert, debug=self.debug
+            self.endpoints["api"],
+            cacert=self.cacert,
+            debug=self.debug,
+            basic_auth=self.basic_auth,
         )
 
         # Service Registry
@@ -246,6 +327,7 @@ def __init__(
             self.endpoints["api"],
             cacert=self.cacert,
             debug=self.debug,
+            basic_auth=self.basic_auth,
         )
 
         self.managers["ServiceRegistryMembers"] = ServiceRegistryMembersManager(
@@ -253,17 +335,23 @@ def __init__(
             self.endpoints["api"],
             cacert=self.cacert,
             debug=self.debug,
+            basic_auth=self.basic_auth,
         )
 
         # RBAC
         self.managers["Role"] = ResourceManager(
-            models.Role, self.endpoints["api"], cacert=self.cacert, debug=self.debug
+            models.Role,
+            self.endpoints["api"],
+            cacert=self.cacert,
+            debug=self.debug,
+            basic_auth=self.basic_auth,
         )
         self.managers["UserRoleAssignment"] = ResourceManager(
             models.UserRoleAssignment,
             self.endpoints["api"],
             cacert=self.cacert,
             debug=self.debug,
+            basic_auth=self.basic_auth,
         )
 
     @add_auth_token_to_kwargs_from_env
@@ -275,7 +363,10 @@ def get_user_info(self, **kwargs):
         """
         url = "/user"
         client = httpclient.HTTPClient(
-            root=self.endpoints["api"], cacert=self.cacert, debug=self.debug
+            root=self.endpoints["api"],
+            cacert=self.cacert,
+            debug=self.debug,
+            basic_auth=self.basic_auth,
         )
         response = client.get(url=url, **kwargs)
 
diff --git a/st2client/st2client/config_parser.py b/st2client/st2client/config_parser.py
index c572e101f0..faac5baa2e 100644
--- a/st2client/st2client/config_parser.py
+++ b/st2client/st2client/config_parser.py
@@ -59,6 +59,11 @@
         "username": {"type": "string", "default": None},
         "password": {"type": "string", "default": None},
         "api_key": {"type": "string", "default": None},
+        "basic_auth": {
+            # Basic auth credentials in username:password notation
+            "type": "string",
+            "default": None,
+        },
     },
     "api": {"url": {"type": "string", "default": None}},
     "auth": {"url": {"type": "string", "default": None}},
diff --git a/st2client/st2client/models/core.py b/st2client/st2client/models/core.py
index f9b6cbf243..412abd99be 100644
--- a/st2client/st2client/models/core.py
+++ b/st2client/st2client/models/core.py
@@ -15,6 +15,9 @@
 
 from __future__ import absolute_import
 
+from typing import Optional
+from typing import Tuple
+
 import os
 import logging
 from functools import wraps
@@ -175,10 +178,30 @@ def __repr__(self):
 
 
 class ResourceManager(object):
-    def __init__(self, resource, endpoint, cacert=None, debug=False):
+    def __init__(
+        self,
+        resource: str,
+        endpoint: str,
+        cacert: Optional[str] = None,
+        debug: bool = False,
+        basic_auth: Optional[Tuple[str, str]] = None,
+    ):
+        """
+        :param resource: Name of the resource to operate on.
+        :param endpoint: API endpoint URL.
+        :param cacert: Optional path to CA cert to use to validate the server side cert.
+        :param debug: True to enable debug mode where additional debug information will be logged.
+        :param basic_auth: Optional additional basic auth credentials in tuple(username, password)
+                           notation.
+        """
         self.resource = resource
+        self.endpoint = endpoint
+        self.cacert = cacert
         self.debug = debug
-        self.client = httpclient.HTTPClient(endpoint, cacert=cacert, debug=debug)
+        self.basic_auth = basic_auth
+        self.client = httpclient.HTTPClient(
+            endpoint, cacert=cacert, debug=debug, basic_auth=basic_auth
+        )
 
     @staticmethod
     def handle_error(response):
@@ -436,11 +459,6 @@ def delete_by_id(self, instance_id, **kwargs):
 
 
 class ActionAliasResourceManager(ResourceManager):
-    def __init__(self, resource, endpoint, cacert=None, debug=False):
-        self.resource = resource
-        self.debug = debug
-        self.client = httpclient.HTTPClient(root=endpoint, cacert=cacert, debug=debug)
-
     @add_auth_token_to_kwargs_from_env
     def match(self, instance, **kwargs):
         url = "/%s/match" % self.resource.get_url_path_name()
@@ -680,11 +698,6 @@ def update(self, instance, **kwargs):
 
 
 class WebhookManager(ResourceManager):
-    def __init__(self, resource, endpoint, cacert=None, debug=False):
-        self.resource = resource
-        self.debug = debug
-        self.client = httpclient.HTTPClient(root=endpoint, cacert=cacert, debug=debug)
-
     @add_auth_token_to_kwargs_from_env
     def post_generic_webhook(self, trigger, payload=None, trace_tag=None, **kwargs):
         url = "/webhooks/st2"
@@ -715,11 +728,16 @@ def match(self, instance, **kwargs):
         )
 
 
-class StreamManager(object):
-    def __init__(self, endpoint, cacert=None, debug=False):
+class StreamManager(ResourceManager):
+    def __init__(self, endpoint, cacert=None, debug=False, basic_auth=None):
+        super(StreamManager, self).__init__(
+            resource=None,
+            endpoint=endpoint,
+            cacert=cacert,
+            debug=debug,
+            basic_auth=basic_auth,
+        )
         self._url = httpclient.get_url_without_trailing_slash(endpoint) + "/stream"
-        self.debug = debug
-        self.cacert = cacert
 
     @add_auth_token_to_kwargs_from_env
     def listen(self, events=None, **kwargs):
@@ -752,6 +770,9 @@ def listen(self, events=None, **kwargs):
         if self.cacert is not None:
             request_params["verify"] = self.cacert
 
+        if self.basic_auth:
+            request_params["auth"] = self.basic_auth
+
         query_string = "?" + urllib.parse.urlencode(query_params)
         url = url + query_string
 
@@ -767,12 +788,12 @@ def listen(self, events=None, **kwargs):
 
 
 class WorkflowManager(object):
-    def __init__(self, endpoint, cacert, debug):
+    def __init__(self, endpoint, cacert, debug, basic_auth=None):
         self.debug = debug
         self.cacert = cacert
         self.endpoint = endpoint + "/workflows"
         self.client = httpclient.HTTPClient(
-            root=self.endpoint, cacert=cacert, debug=debug
+            root=self.endpoint, cacert=cacert, debug=debug, basic_auth=basic_auth
         )
 
     @staticmethod
diff --git a/st2client/st2client/shell.py b/st2client/st2client/shell.py
index 5c3a6b9cd8..911795524e 100755
--- a/st2client/st2client/shell.py
+++ b/st2client/st2client/shell.py
@@ -251,6 +251,14 @@ def __init__(self):
             "If this is not provided, then SSL cert will not be verified.",
         )
 
+        self.parser.add_argument(
+            "--basic-auth",
+            action="store",
+            dest="basic_auth",
+            default=None,
+            help="Optional additional basic auth credentials used to authenticate",
+        )
+
         self.parser.add_argument(
             "--config-file",
             action="store",
diff --git a/st2client/st2client/utils/httpclient.py b/st2client/st2client/utils/httpclient.py
index 6af6595ec5..ec7bebdf64 100644
--- a/st2client/st2client/utils/httpclient.py
+++ b/st2client/st2client/utils/httpclient.py
@@ -35,6 +35,34 @@ def decorate(*args, **kwargs):
     return decorate
 
 
+def add_basic_auth_creds_to_kwargs(func):
+    """
+    Add "auth" tuple parameter to the kwargs object which is passed to requests method in case it's
+    present on the HTTPClient object instance.
+    """
+
+    def decorate(*args, **kwargs):
+        # NOTE: When logging in using /v1/auth/tokens API endpoint, "auth" argument will already be
+        # present since basic authentication is used to authenticate against auth service to obtain
+        # a token.
+        #
+        # In such scenarios, we don't pass additional basic auth headers to the server.
+        #
+        # This is not ideal, because it means if additional proxy based http auth is enabled, user
+        # may be able to authenticate against StackStorm auth service and obtain a valid auth token
+        # without using additional basic auth credentials, but all the request of the API operations
+        # on StackStorm API won't work without additional basic auth credentials.
+        if (
+            "auth" not in kwargs
+            and isinstance(args[0], HTTPClient)
+            and getattr(args[0], "basic_auth", None)
+        ):
+            kwargs["auth"] = args[0].basic_auth
+        return func(*args, **kwargs)
+
+    return decorate
+
+
 def add_auth_token_to_headers(func):
     def decorate(*args, **kwargs):
         headers = kwargs.get("headers", dict())
@@ -79,13 +107,15 @@ def get_url_without_trailing_slash(value):
 
 
 class HTTPClient(object):
-    def __init__(self, root, cacert=None, debug=False):
+    def __init__(self, root, cacert=None, debug=False, basic_auth=None):
         self.root = get_url_without_trailing_slash(root)
         self.cacert = cacert
         self.debug = debug
+        self.basic_auth = basic_auth
 
     @add_ssl_verify_to_kwargs
     @add_auth_token_to_headers
+    @add_basic_auth_creds_to_kwargs
     def get(self, url, **kwargs):
         response = requests.get(self.root + url, **kwargs)
         response = self._response_hook(response=response)
@@ -94,6 +124,7 @@ def get(self, url, **kwargs):
     @add_ssl_verify_to_kwargs
     @add_auth_token_to_headers
     @add_json_content_type_to_headers
+    @add_basic_auth_creds_to_kwargs
     def post(self, url, data, **kwargs):
         response = requests.post(self.root + url, json.dumps(data), **kwargs)
         response = self._response_hook(response=response)
@@ -101,6 +132,7 @@ def post(self, url, data, **kwargs):
 
     @add_ssl_verify_to_kwargs
     @add_auth_token_to_headers
+    @add_basic_auth_creds_to_kwargs
     def post_raw(self, url, data, **kwargs):
         response = requests.post(self.root + url, data, **kwargs)
         response = self._response_hook(response=response)
@@ -109,6 +141,7 @@ def post_raw(self, url, data, **kwargs):
     @add_ssl_verify_to_kwargs
     @add_auth_token_to_headers
     @add_json_content_type_to_headers
+    @add_basic_auth_creds_to_kwargs
     def put(self, url, data, **kwargs):
         response = requests.put(self.root + url, json.dumps(data), **kwargs)
         response = self._response_hook(response=response)
@@ -117,6 +150,7 @@ def put(self, url, data, **kwargs):
     @add_ssl_verify_to_kwargs
     @add_auth_token_to_headers
     @add_json_content_type_to_headers
+    @add_basic_auth_creds_to_kwargs
     def patch(self, url, data, **kwargs):
         response = requests.patch(self.root + url, data, **kwargs)
         response = self._response_hook(response=response)
@@ -124,6 +158,7 @@ def patch(self, url, data, **kwargs):
 
     @add_ssl_verify_to_kwargs
     @add_auth_token_to_headers
+    @add_basic_auth_creds_to_kwargs
     def delete(self, url, **kwargs):
         response = requests.delete(self.root + url, **kwargs)
         response = self._response_hook(response=response)
diff --git a/st2client/tests/fixtures/st2rc.full.ini b/st2client/tests/fixtures/st2rc.full.ini
index a4ac10f746..b55bd619d0 100644
--- a/st2client/tests/fixtures/st2rc.full.ini
+++ b/st2client/tests/fixtures/st2rc.full.ini
@@ -10,6 +10,8 @@ cache_token = False
 [credentials]
 username = test1
 password = test1
+api_key = api_key
+basic_auth = user1:pass1
 
 [api]
 url = http://127.0.0.1:9101/v1
diff --git a/st2client/tests/unit/test_client.py b/st2client/tests/unit/test_client.py
index 2d0a380ab5..01d5bb4800 100644
--- a/st2client/tests/unit/test_client.py
+++ b/st2client/tests/unit/test_client.py
@@ -14,14 +14,21 @@
 # limitations under the License.
 
 from __future__ import absolute_import
+
 import os
-import six
+import json
 import logging
 import unittest2
 
+import six
+import mock
+import requests
+
 from st2client import models
 from st2client.client import Client
 
+from tests import base
+
 
 LOG = logging.getLogger(__name__)
 
@@ -65,6 +72,50 @@ def test_default(self):
         self.assertEqual(endpoints["api"], api_url)
         self.assertEqual(endpoints["stream"], stream_url)
 
+    @mock.patch.object(
+        requests,
+        "get",
+        mock.MagicMock(return_value=base.FakeResponse(json.dumps({}), 200, "OK")),
+    )
+    def test_basic_auth_option_success(self):
+        client = Client(basic_auth="username:password")
+        self.assertEqual(client.basic_auth, ("username", "password"))
+
+        self.assertEqual(requests.get.call_count, 0)
+        client.actions.get_all()
+        self.assertEqual(requests.get.call_count, 1)
+
+        requests.get.assert_called_with(
+            "http://127.0.0.1:9101/v1/actions", auth=("username", "password"), params={}
+        )
+
+    @mock.patch.object(
+        requests,
+        "get",
+        mock.MagicMock(return_value=base.FakeResponse(json.dumps({}), 200, "OK")),
+    )
+    def test_basic_auth_option_success_password_with_colon(self):
+        client = Client(basic_auth="username:password:with:colon")
+        self.assertEqual(client.basic_auth, ("username", "password:with:colon"))
+
+        self.assertEqual(requests.get.call_count, 0)
+        client.actions.get_all()
+        self.assertEqual(requests.get.call_count, 1)
+
+        requests.get.assert_called_with(
+            "http://127.0.0.1:9101/v1/actions",
+            auth=("username", "password:with:colon"),
+            params={},
+        )
+
+    def test_basic_auth_option_invalid_notation(self):
+        self.assertRaisesRegex(
+            ValueError,
+            "needs to be in the username:password notation",
+            Client,
+            basic_auth="username_password",
+        )
+
     def test_env(self):
         base_url = "http://www.stackstorm.com"
         api_url = "http://www.st2.com:9101/v1"
diff --git a/st2client/tests/unit/test_config_parser.py b/st2client/tests/unit/test_config_parser.py
index 0fb3e0db8a..eec8694442 100644
--- a/st2client/tests/unit/test_config_parser.py
+++ b/st2client/tests/unit/test_config_parser.py
@@ -64,7 +64,12 @@ def test_parse(self):
                 "silence_schema_output": True,
             },
             "cli": {"debug": True, "cache_token": False, "timezone": "UTC"},
-            "credentials": {"username": "test1", "password": "test1", "api_key": None},
+            "credentials": {
+                "username": "test1",
+                "password": "test1",
+                "api_key": "api_key",
+                "basic_auth": "user1:pass1",
+            },
             "api": {"url": "http://127.0.0.1:9101/v1"},
             "auth": {"url": "http://127.0.0.1:9100/"},
             "stream": {"url": "http://127.0.0.1:9102/v1/stream"},