MB-52782 Use the correct credentials for sink REST requests

Prior to this change, 'cbrecovery' used the source credentials for
operations on both the source and sink.

This resulted in the inability to recover to/from clusters where the
credentials were different.

Change-Id: I05b0315ba3bd5b3a071754192498082516b5d994
Reviewed-on: https://review.couchbase.org/c/couchbase-cli/+/177091
Tested-by: Build Bot <build@couchbase.com>
Reviewed-by: Maksimiljans Januska <maks.januska@couchbase.com>
Well-Formed: Restriction Checker

Author: jamesl33

cbrecovery

@@ -126,8 +126,10 @@ class Recovery(pump_transfer.Transfer):
         sys.stderr.write(f'{total_vbucket} vbuckets recovered with elapsed time {delta:0.2f} seconds\n')
 
     def pre_transfer(self, opts, source, sink):
+        host, port, _, _, path = pump.parse_spec(opts, sink, 8091)
 
-        host, port, user, pwd, path = pump.parse_spec(opts, sink, 8091)
+        # NOTE: Request is going to the destination cluster e.g. the one being recovered. Use the correct credentials.
+        user, pwd = opts.username_dest, opts.password_dest
 
         # retrieve a list of missing vbucket
         cmd = "startRecovery"
@@ -148,9 +150,8 @@ class Recovery(pump_transfer.Transfer):
 
     def post_transfer(self, opts, source, sink, vbucket):
         if not self.sink_bucket:
-            return "Should specify destionation bucket for restore"
+            return "Should specify destination bucket for restore", None
 
-        host, port, user, pwd, path = pump.parse_spec(opts, sink, 8091)
         if opts.dry_run:
             cmd = "stopRecovery"
             reason_msg = "Stop recovery"
@@ -161,6 +162,12 @@ class Recovery(pump_transfer.Transfer):
             params = urllib.parse.urlencode({"vbucket": vbucket})
         else:
             params = None
+
+        host, port, _, _, path = pump.parse_spec(opts, sink, 8091)
+
+        # NOTE: Request is going to the destination cluster e.g. the one being recovered. Use the correct credentials.
+        user, pwd = opts.username_dest, opts.password_dest
+
         url = f'/pools/default/buckets/{self.sink_bucket}/controller/{cmd}?recovery_uuid={self.recovery_uuid}'
         post_headers = {"Content-type": "application/x-www-form-urlencoded"}
         err, _, _ = pump.rest_request(host, int(port), user, pwd, False, url, method='POST', body=params,

test/mock_server.py

@@ -16,6 +16,7 @@
 from cryptography.hazmat.primitives import hashes, serialization
 from cryptography.hazmat.primitives.asymmetric import rsa
 from cryptography.x509.oid import NameOID
+from requests.auth import HTTPBasicAuth
 
 
 # generate_self_signed_cert generates a key/self signed certificate pair which will be written to key.pem/cert.pem in
@@ -65,6 +66,10 @@ class RequestHandler(BaseHTTPRequestHandler):
     def do_GET(self):
         parsed = urlparse(self.path)
         self.server.rest_server.trace.append(f'GET:{parsed.path}')
+
+        if not self.authenticated():
+            return
+
         for (endpoint, fns) in endpoints:
             if re.search(endpoint, parsed.path) is not None and 'GET' in fns:
                 return self.handle_fn(fns['GET'], parsed.path, parsed.query)
@@ -74,6 +79,10 @@ def do_GET(self):
     def do_POST(self):
         parsed = urlparse(self.path)
         self.server.rest_server.trace.append(f'POST:{parsed.path}')
+
+        if not self.authenticated():
+            return
+
         for (endpoint, fns) in endpoints:
             if re.search(endpoint, parsed.path) is not None and 'POST' in fns:
                 return self.handle_fn(fns['POST'], parsed.path, parsed.query)
@@ -83,6 +92,10 @@ def do_POST(self):
     def do_PATCH(self):
         parsed = urlparse(self.path)
         self.server.rest_server.trace.append(f'PATCH:{parsed.path}')
+
+        if not self.authenticated():
+            return
+
         for (endpoint, fns) in endpoints:
             if re.search(endpoint, parsed.path) is not None and 'PATCH' in fns:
                 return self.handle_fn(fns['PATCH'], parsed.path)
@@ -92,6 +105,10 @@ def do_PATCH(self):
     def do_PUT(self):
         parsed = urlparse(self.path)
         self.server.rest_server.trace.append(f'PUT:{parsed.path}')
+
+        if not self.authenticated():
+            return
+
         for (endpoint, fns) in endpoints:
             if re.search(endpoint, parsed.path) is not None and 'PUT' in fns:
                 return self.handle_fn(fns['PUT'], parsed.path)
@@ -104,11 +121,23 @@ def do_DELETE(self):
         if parsed.query:
             self.server.rest_server.queries.append(parsed.query)
 
+        if not self.authenticated():
+            return
+
         for (endpoint, fns) in endpoints:
             if re.search(endpoint, parsed.path) is not None and 'DELETE' in fns:
                 return self.handle_fn(fns['DELETE'], parsed.path, parsed.query)
         self.not_found()
 
+    def authenticated(self):
+        # Username/password == Administrator/asdasd
+        if self.headers.get('Authorization') == f"Basic QWRtaW5pc3RyYXRvcjphc2Rhc2Q=":
+            return True
+
+        self.send_response(401)
+
+        return False
+
     def not_found(self):
         self.send_response(404)
         self.finish()
@@ -117,7 +146,7 @@ def handle_fn(self, fn, path, params=None):
         content_len = int(self.headers.get('content-length', 0))
         post_body = self.rfile.read(content_len).decode('utf-8')
 
-        if self.headers.get('Content-Type', 'application/x-www-form-urlencoded') == 'application/json':
+        if post_body and self.headers.get('Content-Type', 'application/x-www-form-urlencoded') == 'application/json':
             # to help with verifying later on we are going to load this json and then dump it again but with sorted keys
             # to ensure stable serializing so then we can do string comparison on the results
             self.server.rest_server.rest_params.append(json.dumps(json.loads(post_body), sort_keys=True))
@@ -215,7 +244,7 @@ def shutdown(self):
 
     def _close(self, url, server, t):
         try:
-            requests.get(f'{url}/close', timeout=0.2, verify=False)
+            requests.get(f'{url}/close', auth=HTTPBasicAuth('Administrator', 'asdasd'), timeout=0.2, verify=False)
         except Exception:
             pass
 
@@ -541,6 +570,9 @@ def export_eventing_functions(rest_params=None, server_args=None, path="", endpo
     (r'/controller/uploadClusterCA', {'POST': do_nothing}),
     (r'/controller/regenerateCertificate', {'POST': get_by_path}),
 
+    # cbrecovery
+    (r'/pools/default/buckets/\w+/controller/startRecovery', {'POST': get_by_path}),
+    (r'/pools/default/buckets/\w+/controller/commitVBucket', {'POST': get_by_path}),
 
     # index api
     (r'/getIndexMetadata$', {'GET': get_by_path}),

test/test_cbrecovery.py

@@ -0,0 +1,84 @@
+#!/usr/bin/env python3
+
+# Copyright 2022 Couchbase Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#        http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import imp
+import json
+import os
+import unittest
+from optparse import Values
+
+from mock_server import MockRESTServer, generate_self_signed_cert
+
+cbrecovery = imp.load_source('cbrecovery', './cbrecovery')
+
+host = '127.0.0.1'
+port = 6789
+
+# setUpModule will generate new certificates for the mock HTTPS server used during unit testing.
+#
+# NOTE: We don't remove the key/cert once generated since they're included in the '.gitignore' file.
+
+
+def setUpModule():
+    generate_self_signed_cert(os.path.dirname(os.path.abspath(__file__)))
+
+
+class RecoveryTest(unittest.TestCase):
+    def test_pre_transfer(self):
+        server = MockRESTServer(host, port)
+        server.set_args({"/pools/default/buckets/bucket/controller/startRecovery": {}})
+        server.run()
+
+        recovery = cbrecovery.Recovery()
+        recovery.sink_bucket = "bucket"
+
+        opts = Values({
+            "username": "",
+            "password": "",
+            # Assert that we use these credentials, and not the empty source credentials
+            "username_dest": "Administrator",
+            "password_dest": "asdasd",
+            "ssl": False,
+        })
+
+        err, _ = recovery.pre_transfer(opts, "", f"http://{host}:{port}")
+        self.assertEqual("Missing recovery map from response", err)
+        self.assertIn('POST:/pools/default/buckets/bucket/controller/startRecovery', server.trace)
+
+        server.shutdown()
+
+    def test_post_transfer(self):
+        server = MockRESTServer(host, port)
+        server.run()
+
+        recovery = cbrecovery.Recovery()
+        recovery.sink_bucket = "bucket"
+
+        opts = Values({
+            "username": "",
+            "password": "",
+            # Assert that we use these credentials, and not the empty source credentials
+            "username_dest": "Administrator",
+            "password_dest": "asdasd",
+            "dry_run": False,
+            "ssl": False,
+        })
+
+        err, _ = recovery.post_transfer(opts, "", f"http://{host}:{port}", 0)
+        self.assertIn('POST:/pools/default/buckets/bucket/controller/commitVBucket', server.trace)
+        self.assertIsNone(err)
+
+        server.shutdown()

test/test_pumps.py

@@ -546,7 +546,8 @@ def test_provide_index(self):
                        '/getIndexMetadata': {'result': 'INDEX'}}
         server.set_args(server_args)
         server.run()
-        opts = Ditto({'username': 'admin', 'password': 'asdasd', 'ssl': False, 'no_ssl_verify': True, 'cacert': None})
+        opts = Ditto({'username': 'Administrator', 'password': 'asdasd',
+                     'ssl': False, 'no_ssl_verify': True, 'cacert': None})
         error, index = DCPStreamSource.provide_index(opts, 'http://127.0.0.1:9211', {'name': 'default'}, None)
         server.shutdown()
         self.assertEqual(0, error, 'Got unexpected error: {}'.format(error))
@@ -562,7 +563,8 @@ def test_provide_fts_index(self):
                        }}}}
         server.set_args(server_args)
         server.run()
-        opts = Ditto({'username': 'admin', 'password': 'asdasd', 'ssl': False, 'no_ssl_verify': True, 'cacert': None})
+        opts = Ditto({'username': 'Administrator', 'password': 'asdasd',
+                     'ssl': False, 'no_ssl_verify': True, 'cacert': None})
         error, index = DCPStreamSource.provide_fts_index(opts, 'http://127.0.0.1:9211', {'name': 'default'}, None)
         server.shutdown()
         self.assertEqual(0, error, 'Got unexpected error: {}'.format(error))
@@ -1248,9 +1250,9 @@ def handle_sasl_auth_plain(self, data, req, log=False):
                      cas=cas)
             return
         user_pass = body[keylen:]
-        if user_pass != b'\x00admin\x00asdasd':
+        if user_pass != b'\x00Administrator\x00asdasd':
             self.failed = True
-            self.reason = "Received {} expected {}".format(user_pass, b'\x00admin\x00asdasd')
+            self.reason = "Received {} expected {}".format(user_pass, b'\x00Administrator\x00asdasd')
             self.res(req, cbcs.CMD_SASL_AUTH, errcode=cbcs.ERR_EINTERNAL, body=self.reason.encode(), opaque=opaque,
                      cas=cas)
             return
@@ -1339,7 +1341,7 @@ class TestMemcachedClient(unittest.TestCase):
     def setUp(self):
         self.server = MockMemcachedServer(debug=False)
         self.host, self.port = self.server.get_host_address()
-        self.helper_class = MCHelper({'response': b'res_data', 'user': 'admin', 'pass': 'asdasd'})
+        self.helper_class = MCHelper({'response': b'res_data', 'user': 'Administrator', 'pass': 'asdasd'})
 
     def tearDown(self):
         self.server.stop()
@@ -1397,7 +1399,7 @@ def test_sasl_auth_plain(self):
 
         try:
             client = MemcachedClient(self.host, self.port)
-            opaque_o, cas_o, data = client.sasl_auth_plain('admin', 'asdasd')
+            opaque_o, cas_o, data = client.sasl_auth_plain('Administrator', 'asdasd')
         except Exception as e:
             client.close()
             self.server.stop()