[lldb/Plugins] Add memory writing capabilities to Scripted Process

[mib]

Summary

This patch adds memory writing capabilities to the Scripted Process plugin.

This allows to user to get a target address and a memory buffer on the
python scripted process implementation that the user can make processing
on before performing the actual write.

This will also be used to write trap instruction to a real process
memory to set a breakpoint.

Signed-off-by: Med Ismail Bennani <[e-mail]>

[mib]

Reformat

[mib]

Fix typo

[bulbazord]

lldb/source/Plugins/Process/scripted/ScriptedProcess.cpp

310–311

This function is added in this patch. Where is it used?

325

Should this not return bytes_written?

[mib]

lldb/source/Plugins/Process/scripted/ScriptedProcess.cpp

310–311

Plugin::DoX functions are usually implemented by plugins and gets called by the base component Base::X method. In this case, this will get called by the Process::WriteMemory method.

325

+1

[mib]

Address @bulbazord comments.

[bulbazord]

LGTM

[mib]

Rebase

[mib]

Landed in f190ec688270

lldb/examples/python/scripted_process/scripted_process.py

def read_memory_at_address(self, addr, size, error):
             error (lldb.SBError): Error object.
 
         Returns:
             lldb.SBData: An `lldb.SBData` buffer with the target byte size and
                 byte order storing the memory read.
         """
         pass
 
+    def write_memory_at_address(self, addr, data, error):
+        """ Write a buffer to the scripted process memory.
+
+        Args:
+            addr (int): Address from which we should start reading.
+            data (lldb.SBData): An `lldb.SBData` buffer to write to the
+                process memory.
+            error (lldb.SBError): Error object.
+
+        Returns:
+            size (int): Size of the memory to read.
+        """
+        error.SetErrorString("$s doesn't support memory writting." % self.__class__.__name__)
+        return 0
+
     def get_loaded_images(self):
         """ Get the list of loaded images for the scripted process.
 
         ```
         scripted_image = {
             uuid = "c6ea2b64-f77c-3d27-9528-74f507b9078b",
             path = "/usr/lib/dyld"
             load_addr = 0xbadc0ffee

lldb/include/lldb/Interpreter/ScriptedProcessInterface.h

public:
   }
 
   virtual StructuredData::DictionarySP GetThreadsInfo() { return {}; }
 
   virtual lldb::DataExtractorSP
   ReadMemoryAtAddress(lldb::addr_t address, size_t size, Status &error) {
     return {};
   }
   
+  virtual size_t
+  WriteMemoryAtAddress(lldb::addr_t addr, lldb::DataExtractorSP data_sp, Status &error) {
+    return LLDB_INVALID_OFFSET;
+  };
+
   virtual StructuredData::ArraySP GetLoadedImages() { return {}; }
 
   virtual lldb::pid_t GetProcessID() { return LLDB_INVALID_PROCESS_ID; }
 
   virtual bool IsAlive() { return true; }
 
   virtual StructuredData::DictionarySP GetMetadata() { return {}; }
 

lldb/source/Plugins/Process/scripted/ScriptedProcess.h

public:
   Status DoDestroy() override;
 
   void RefreshStateAfterStop() override;
 
   bool IsAlive() override;
 
   size_t DoReadMemory(lldb::addr_t addr, void *buf, size_t size,
                       Status &error) override;
   
+  size_t DoWriteMemory(lldb::addr_t vm_addr, const void *buf,
+                       size_t size, Status &error) override;
+
   ArchSpec GetArchitecture();
 
   Status
   GetMemoryRegions(lldb_private::MemoryRegionInfos &region_list) override;
 
   bool GetProcessInfo(ProcessInstanceInfo &info) override;
 
   lldb_private::StructuredData::ObjectSP

lldb/source/Plugins/Process/scripted/ScriptedProcess.cpp

size_t ScriptedProcess::DoReadMemory(lldb::addr_t addr, void *buf, size_t size,
 
   if (!bytes_copied || bytes_copied == LLDB_INVALID_OFFSET)
     return ScriptedInterface::ErrorWithMessage<size_t>(
         LLVM_PRETTY_FUNCTION, "Failed to copy read memory to buffer.", error);
 
   return size;
 }
 
+size_t ScriptedProcess::DoWriteMemory(lldb::addr_t vm_addr, const void *buf,
+                                      size_t size, Status &error) {
+  lldb::DataExtractorSP data_extractor_sp = std::make_shared<DataExtractor>(buf, size, GetByteOrder(), GetAddressByteSize());
+  
+  if (!data_extractor_sp || !data_extractor_sp->GetByteSize())
+    return 0;
+  
+  size_t bytes_written = GetInterface().WriteMemoryAtAddress(vm_addr, data_extractor_sp, error);
+  
+  if (!bytes_written || bytes_written == LLDB_INVALID_OFFSET)
+    return ScriptedInterface::ErrorWithMessage<size_t>(
+        LLVM_PRETTY_FUNCTION, "Failed to copy write buffer to memory.", error);
+
+  return size;
+}
+
 ArchSpec ScriptedProcess::GetArchitecture() {
   return GetTarget().GetArchitecture();
 }
 
 Status ScriptedProcess::DoGetMemoryRegionInfo(lldb::addr_t load_addr,
                                               MemoryRegionInfo &region) {
   Status error;
   if (auto region_or_err =

lldb/source/Plugins/ScriptInterpreter/Python/ScriptedProcessPythonInterface.h

public:
   GetMemoryRegionContainingAddress(lldb::addr_t address,
                                    Status &error) override;
 
   StructuredData::DictionarySP GetThreadsInfo() override;
 
   lldb::DataExtractorSP ReadMemoryAtAddress(lldb::addr_t address, size_t size,
                                             Status &error) override;
                                          
+  size_t WriteMemoryAtAddress(lldb::addr_t addr, lldb::DataExtractorSP data_sp, Status &error) override;
 
   StructuredData::ArraySP GetLoadedImages() override;
 
   lldb::pid_t GetProcessID() override;
 
   bool IsAlive() override;
 
   StructuredData::DictionarySP GetMetadata() override;

lldb/source/Plugins/ScriptInterpreter/Python/ScriptedProcessPythonInterface.cpp

lldb::DataExtractorSP ScriptedProcessPythonInterface::ReadMemoryAtAddress(
 
   // If there was an error on the python call, surface it to the user.
   if (py_error.Fail())
     error = py_error;
 
   return data_sp;
 }
 
+size_t ScriptedProcessPythonInterface::WriteMemoryAtAddress(lldb::addr_t addr, lldb::DataExtractorSP data_sp, Status &error) {
+  Status py_error;
+  StructuredData::ObjectSP obj = Dispatch("write_memory_at_address", py_error, addr, data_sp, error);
+
+  if (!CheckStructuredDataObject(LLVM_PRETTY_FUNCTION, obj, error))
+    return LLDB_INVALID_OFFSET;
+
+  // If there was an error on the python call, surface it to the user.
+  if (py_error.Fail())
+    error = py_error;
+
+  return obj->GetIntegerValue(LLDB_INVALID_OFFSET);
+}
+
 StructuredData::ArraySP ScriptedProcessPythonInterface::GetLoadedImages() {
   Status error;
   StructuredData::ArraySP array =
       Dispatch<StructuredData::ArraySP>("get_loaded_images", error);
 
   if (!CheckStructuredDataObject(LLVM_PRETTY_FUNCTION, array, error))
     return {};
 

lldb/test/API/functionalities/scripted_process/TestScriptedProcess.py

def test_scripted_process_and_scripted_thread(self):
 
         error = lldb.SBError()
         process_1 = target_1.Attach(attach_info, error)
         self.assertTrue(process_1 and process_1.IsValid(), PROCESS_IS_VALID)
         self.assertEqual(process_1.GetProcessID(), 42)
         self.assertEqual(process_1.GetNumThreads(), 1)
 
         # ... then try reading from target #1 process ...
-        addr = 0x500000000
         message = "Hello, target 1"
         buff = process_1.ReadCStringFromMemory(addr, len(message) + 1, error)
         self.assertSuccess(error)
         self.assertEqual(buff, message)
 
         # ... now, reading again from target #0 process to make sure the call
         # gets dispatched to the right target.
-        addr = 0x500000000
         message = "Hello, target 0"
         buff = process_0.ReadCStringFromMemory(addr, len(message) + 1, error)
         self.assertSuccess(error)
         self.assertEqual(buff, message)
 
+        # Let's write some memory.
+        message = "Hello, world!"
+        bytes_written = process_0.WriteCStringToMemory(addr, message, error)
+        self.assertSuccess(error)
+        self.assertEqual(bytes_written, len(message) + 1)
+
+        # ... and check if that memory was saved properly.
+        buff = process_0.ReadCStringFromMemory(addr, len(message) + 1, error)
+        self.assertSuccess(error)
+        self.assertEqual(buff, message)
+
         thread = process_0.GetSelectedThread()
         self.assertTrue(thread, "Invalid thread.")
         self.assertEqual(thread.GetThreadID(), 0x19)
         self.assertEqual(thread.GetName(), "DummyScriptedThread.thread-1")
         self.assertStopReason(thread.GetStopReason(), lldb.eStopReasonSignal)
 
         self.assertGreater(thread.GetNumFrames(), 0)
 

lldb/test/API/functionalities/scripted_process/dummy_scripted_process.py

 import os,struct, signal
 
 from typing import Any, Dict
 
 import lldb
 from lldb.plugins.scripted_process import ScriptedProcess
 from lldb.plugins.scripted_process import ScriptedThread
 
 class DummyScriptedProcess(ScriptedProcess):
+    memory = None
+
     def __init__(self, exe_ctx: lldb.SBExecutionContext, args : lldb.SBStructuredData):
         super().__init__(exe_ctx, args)
         self.threads[0] = DummyScriptedThread(self, None)
-
-    def read_memory_at_address(self, addr: int, size: int, error: lldb.SBError) -> lldb.SBData:
+        self.memory = {}
+        addr = 0x500000000
         debugger = self.target.GetDebugger()
         index = debugger.GetIndexOfTarget(self.target)
+        self.memory[addr] = "Hello, target " + str(index)
+
+    def read_memory_at_address(self, addr: int, size: int, error: lldb.SBError) -> lldb.SBData:
         data = lldb.SBData().CreateDataFromCString(
                                     self.target.GetByteOrder(),
                                     self.target.GetCodeByteSize(),
-                                    "Hello, target " + str(index))
+                                    self.memory[addr])
 
         return data
 
+    def write_memory_at_address(self, addr, data, error):
+        self.memory[addr] = data.GetString(error, 0)
+        return len(self.memory[addr])
+
     def get_loaded_images(self):
         return self.loaded_images
 
     def get_process_id(self) -> int:
         return 42
 
     def should_stop(self) -> bool:
         return True