/ usr / lib / python3 / dist-packages / samba / tests / dcerpc / [ HOME SHELL ]

Name	Size	Permission	Action
__pycache__	[ DIR ]	drwxr-xr-x
__init__.py	795 B	-rw-r--r--
array.py	6.1 KB	-rw-r--r--
bare.py	2.7 KB	-rw-r--r--
binding.py	3.88 KB	-rw-r--r--
createtrustrelax.py	4.99 KB	-rw-r--r--
dnsserver.py	65.28 KB	-rw-r--r--
integer.py	7.47 KB	-rw-r--r--
lsa.py	12.28 KB	-rw-r--r--
mdssvc.py	6.51 KB	-rw-r--r--
misc.py	3.35 KB	-rw-r--r--
raw_protocol.py	351.73 KB	-rw-r--r--
raw_testcase.py	46.31 KB	-rw-r--r--
registry.py	1.9 KB	-rw-r--r--
rpc_talloc.py	2.82 KB	-rw-r--r--
rpcecho.py	2.36 KB	-rw-r--r--
sam.py	30.41 KB	-rw-r--r--
samr_change_password.py	6.36 KB	-rw-r--r--
srvsvc.py	2.47 KB	-rw-r--r--
string_tests.py	4.29 KB	-rw-r--r--
testrpc.py	5.16 KB	-rw-r--r--
unix.py	1.45 KB	-rw-r--r--

Code Editor : mdssvc.py

#
# Unix SMB/CIFS implementation.
# Copyright Ralph Boehme <slow@samba.org> 2019
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.
#

"""Tests for samba.dcerpc.mdssvc"""

import os
import time
import threading
import logging
import json
from http.server import HTTPServer, BaseHTTPRequestHandler
from samba.dcerpc import mdssvc
from samba.tests import RpcInterfaceTestCase
from samba.samba3 import mdscli
from samba.logger import get_samba_logger

logger = get_samba_logger(name=__name__)

testfiles = [
    "foo",
    "bar",
    "x+x",
    "x*x",
    "x=x",
    "x'x",
    "x?x",
    "x\"x",
    "x\\x",
    "x(x",
    "x x",
]

class MdssvcHTTPRequestHandler(BaseHTTPRequestHandler):
    def do_POST(self):
        content_length = int(self.headers['content-length'])
        body = self.rfile.read(content_length)

actual_json = json.loads((body))
        expected_json = json.loads(self.server.json_in)

if actual_json != expected_json:
            logger.error("Bad request, expected:\n%s\nGot:\n%s\n" % (expected_json, actual_json))
            self.send_error(400,
                            "Bad request",
                            "Expected: %s\n"
                            "Got: %s\n" %
                            (expected_json, actual_json))
            return

resp = bytes(self.server.json_out, encoding="utf-8")

self.send_response(200)
        self.send_header('content-type', 'application/json; charset=UTF-8')
        self.send_header('content-length', len(resp))
        self.end_headers()
        self.wfile.write(resp)

class MdssvcTests(RpcInterfaceTestCase):

def setUp(self):
        super(MdssvcTests, self).setUp()

self.pipe = mdssvc.mdssvc('ncacn_np:fileserver[/pipe/mdssvc]', self.get_loadparm())

self.server = HTTPServer(('10.53.57.35', 8080),
                                 MdssvcHTTPRequestHandler,
                                 bind_and_activate=False)

self.t = threading.Thread(target=MdssvcTests.http_server, args=(self,))
        self.t.setDaemon(True)
        self.t.start()
        self.sharepath = os.environ["LOCAL_PATH"]
        time.sleep(1)

conn = mdscli.conn(self.pipe, 'spotlight', '/foo')
        self.fakepath = conn.sharepath()
        conn.disconnect(self.pipe)

for file in testfiles:
            f = open("%s/%s" % (self.sharepath, file), "w")
            f.close()

def tearDown(self):
        super(RpcInterfaceTestCase, self).tearDown()
        for file in testfiles:
            os.remove("%s/%s" % (self.sharepath, file))

def http_server(self):
        self.server.server_bind()
        self.server.server_activate()
        self.server.serve_forever()

def run_test(self, query, expect, json_in, json_out):
        self.server.json_in = json_in.replace("%BASEPATH%", self.sharepath)
        self.server.json_out = json_out.replace("%BASEPATH%", self.sharepath)

self.conn = mdscli.conn(self.pipe, 'spotlight', '/foo')
        search = self.conn.search(self.pipe, query, self.fakepath)

# Give it some time, the get_results() below returns immediately
        # what's available, so if we ask to soon, we might get back no results
        # as the server is still processing the request
        time.sleep(1)

results = search.get_results(self.pipe)
        self.assertEqual(results, expect)

search.close(self.pipe)
        self.conn.disconnect(self.pipe)

def test_mdscli_search(self):
        exp_json_query = r'''{
          "from": 0, "size": 100, "_source": ["path.real"],
          "query": {
            "query_string": {
              "query": "(samba*) AND path.real.fulltext:\"%BASEPATH%\""
            }
          }
        }'''
        fake_json_response = '''{
          "hits" : {
            "total" : { "value" : 2},
            "hits" : [
              {"_source" : {"path" : {"real" : "%BASEPATH%/foo"}}},
              {"_source" : {"path" : {"real" : "%BASEPATH%/bar"}}}
            ]
          }
        }'''
        exp_results = ["foo", "bar"]
        self.run_test('*=="samba*"', exp_results, exp_json_query, fake_json_response)

def test_mdscli_search_escapes(self):
        sl_query = (
            r'kMDItemFSName=="x+x"||'
            r'kMDItemFSName=="x\*x"||'
            r'kMDItemFSName=="x=x"||'
            'kMDItemFSName=="x\'x"||'
            r'kMDItemFSName=="x?x"||'
            r'kMDItemFSName=="x x"||'
            r'kMDItemFSName=="x(x"||'
            r'kMDItemFSName=="x\"x"||'
            r'kMDItemFSName=="x\\x"'
        )
        exp_json_query = r'''{
          "from": 0, "size": 100, "_source": ["path.real"],
          "query": {
            "query_string": {
              "query": "(file.filename:x\\+x OR file.filename:x\\*x OR file.filename:x=x OR file.filename:x'x OR file.filename:x\\?x OR file.filename:x\\ x OR file.filename:x\\(x OR file.filename:x\\\"x OR file.filename:x\\\\x) AND path.real.fulltext:\"%BASEPATH%\""
            }
          }
        }'''
        fake_json_response = r'''{
          "hits" : {
            "total" : {"value" : 2},
            "hits" : [
              {"_source" : {"path" : {"real" : "%BASEPATH%/x+x"}}},
              {"_source" : {"path" : {"real" : "%BASEPATH%/x*x"}}},
              {"_source" : {"path" : {"real" : "%BASEPATH%/x=x"}}},
              {"_source" : {"path" : {"real" : "%BASEPATH%/x'x"}}},
              {"_source" : {"path" : {"real" : "%BASEPATH%/x?x"}}},
              {"_source" : {"path" : {"real" : "%BASEPATH%/x x"}}},
              {"_source" : {"path" : {"real" : "%BASEPATH%/x(x"}}},
              {"_source" : {"path" : {"real" : "%BASEPATH%/x\"x"}}},
              {"_source" : {"path" : {"real" : "%BASEPATH%/x\\x"}}}
            ]
          }
        }'''
        exp_results = [
            r"x+x",
            r"x*x",
            r"x=x",
            r"x'x",
            r"x?x",
            r"x x",
            r"x(x",
            "x\"x",
            r"x\x",
        ]
        self.run_test(sl_query, exp_results, exp_json_query, fake_json_response)

Urban Research Lab

Urban Research Lab

The Urban Research Lab (URL) at IIIT-Delhi was established in 2020, during a time when the pandemic brought the complexities of urban life more visible than ever. What started as an initiative by faculty and students has grown into a collaborative space for critically exploring and reimagining the urban questions through research, dialogue, and interdisciplinary engagement.