Add automatic tests (#78)

* feat: add tests using impacket

* feat: add PR action

.github/workflows/build_preview.yml

@@ -43,10 +43,9 @@ jobs:
         uses: actions/upload-artifact@v2
         with:
           name: linux-arm
-          path: packages/linux/arm64/*
+          path: packages/linux/arm64/* 
   build_windows:
     runs-on: windows-2019
-
     steps:
     - uses: actions/checkout@v2
     - name: Setup .NET
@@ -72,6 +71,27 @@ jobs:
       with:
         name: windows-x64
         path: packages\windows/x64\*
-
-
-
+  test_on_linux:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v2
+      - name: Setup .NET
+        uses: actions/setup-dotnet@v1
+        with:
+          dotnet-version: 5.0.x
+      - name: Restore dependencies
+        run: dotnet restore
+      - name: Build linux x64
+        run: dotnet publish -c Release --self-contained -r linux-x64 -o packages/linux/amd64 -p:PublishSingleFile=true -p:PublishTrimmed=true -p:InvariantGlobalization=true -p:DebugType=None -p:DebugSymbols=false -p:VersionSuffix=pr$(echo $GITHUB_REF | awk 'BEGIN { FS = "/" } ; { print $3 }')
+      - name: install test dependencies
+        run: |
+          sudo pip install impacket pytest
+      - run: |
+          sudo cp packages/linux/amd64/SMBeagle /bin/smbeagle
+          sudo chmod +x /bin/smbeagle
+          sudo mkdir /empty_dir
+      - name: run pytest
+        run: |
+          cd tests
+          sudo pytest -v

tests/Dockerfile

@@ -0,0 +1,8 @@
+FROM punksecurity/smbeagle
+RUN apt update && apt install python3 python3-pip -y
+RUN pip install impacket pytest
+RUN mkdir /empty_dir
+WORKDIR /tests/
+COPY tests/* .
+ENTRYPOINT [ "" ]
+CMD ["pytest", "-v"]

tests/tests/helpers.py

@@ -0,0 +1,69 @@
+from impacket import smbserver
+import multiprocessing
+import os
+import subprocess
+from time import sleep
+import csv
+import shutil
+import uuid
+
+
+def __setupSMB(address, dir, SMB2 = True):
+    os.chdir("/empty_dir")
+    server = smbserver.SimpleSMBServer(listenAddress=address, listenPort=445)
+    server.addShare("share", dir, "")
+    server.addCredential("test", 1200, "9FD78381EC915F1AAAD3B435B51404EE", "25EDEDFF26CB970623DDA4733227A3F7")
+    server.setSMB2Support(SMB2)
+    server.start()
+
+def setupSMB(address, dir):
+    process = multiprocessing.Process(target=__setupSMB, args=[address, dir])
+    process.start()
+    return process
+
+class SMB(object):
+    def __init__(self, address = "0.0.0.0", dir_structure = ["fileA", "fileB"]):
+        self.address = address
+        self.dir_structure = dir_structure
+        self.dir = f"/{uuid.uuid4().hex}"
+    def __enter__(self):
+        self.smb = setupSMB(self.address, self.dir)
+        os.mkdir(self.dir)
+        self.populate_dir(self.dir, self.dir_structure)
+    def populate_dir(self, dir, dir_structure):
+        for item in dir_structure:
+            if type(item) != type( () ) and type(item) != type(""):
+                raise ValueError("Directory should be list of strings and tuples")
+            if type(item) == type( () ):
+                #type tuple, so create folder and then parse that structure
+                os.mkdir(f"{dir}{os.sep}{item[0]}")
+                self.populate_dir(f"{dir}{os.sep}{item[0]}", item[1])
+            else:
+                # type string, so make the file
+                open(f"{dir}{os.sep}{item}", 'a').close()
+
+    def __exit__(self, *args, **kwargs):
+        self.smb.kill()
+        sleep(1)
+        shutil.rmtree(self.dir)
+        #self.smb.close()
+
+def runSMBeagle(*args, print_out=True):
+    run = subprocess.run(["smbeagle",*args], stdout = subprocess.PIPE, universal_newlines=True)
+    if print_out:
+        print(run.stdout)
+    return run.stdout
+
+def runSMBeagleToCSV(*args):
+    return runSMBeagle("-c","out.csv",*args)
+
+def runSMBeagleToCSVWithAuth(*args):
+    return runSMBeagleToCSV("-u","test", "-p", "goose", *args)
+
+def runSMBeagleToCSVWithAuthAndReturnResults(*args):
+    print(runSMBeagleToCSVWithAuth(*args))
+    with open('out.csv', newline='') as csvfile:
+        results = list(csv.DictReader(csvfile, delimiter=',', quotechar='"'))
+        for result in results:
+            print(result)
+        return results

tests/tests/test_010_options.py

@@ -0,0 +1,41 @@
+from helpers import *
+
+username_or_password_missing_error = "ERROR: Username and Password required on none Windows platforms" 
+def test_username_and_password_required_on_linux():
+    assert username_or_password_missing_error in runSMBeagleToCSV()
+def test_password_required_on_linux():
+    assert username_or_password_missing_error in runSMBeagleToCSV("-p","goose")
+def test_username_required_on_linux():
+    assert username_or_password_missing_error in runSMBeagleToCSV("-u","goose")
+def test_username_and_password_accepted():
+    assert username_or_password_missing_error not in runSMBeagleToCSV("-u","goose", "-p", "goose")
+def test_long_username_accepted():
+    assert username_or_password_missing_error not in runSMBeagleToCSV("--username","goose", "-p", "goose")
+def test_long_password_accepted():
+    assert username_or_password_missing_error not in runSMBeagleToCSV("-u","goose", "--password", "goose")
+
+output_required_error = "At least one option from group 'output' (c, csv-file, e, elasticsearch-host)"
+def test_csv_or_elasticsearch_required():
+    assert output_required_error in runSMBeagle()
+def test_short_csv_accepted():
+    assert output_required_error not in runSMBeagle("-c","out.csv")
+def test_long_csv_accepted():
+    assert output_required_error not in runSMBeagle("--csv-file","out.csv")
+def test_short_elasticsearch_accepted():
+    assert output_required_error not in runSMBeagle("-e","elasticsearch")
+def test_long_elasticsearch_accepted():
+    assert output_required_error not in runSMBeagle("--elasticsearch-host","elasticsearch")
+
+
+def test_manual_host_accepted():
+    assert "127.0.0.2" in runSMBeagleToCSVWithAuth("-h", "127.0.0.2")
+def test_multiple_manual_host_accepted():
+    output = runSMBeagleToCSVWithAuth("-h", "127.0.0.2", "127.0.0.3")
+    assert "127.0.0.2" in output and "127.0.0.3" in output
+
+def test_manual_network_accepted():
+    output = runSMBeagleToCSVWithAuth("-n", "127.0.0.0/24")
+    assert "127.0.0.0/24" in output
+def test_multiple_manual_network_accepted():
+    output = runSMBeagleToCSVWithAuth("-n", "127.0.0.0/24", "127.0.1.0/24")
+    assert "127.0.0.0/24" in output and "127.0.1.0/24" in output

tests/tests/test_020_tcp_scans.py

@@ -0,0 +1,61 @@
+from helpers import *
+
+smb_reachable_message = "we have {} hosts with reachable SMB services"
+
+no_smb_service_discovered_message = smb_reachable_message.format(0)
+one_smb_service_discovered_message = smb_reachable_message.format(1)
+two_smb_service_discovered_message = smb_reachable_message.format(2)
+three_smb_service_discovered_message = smb_reachable_message.format(3)
+four_smb_service_discovered_message = smb_reachable_message.format(4)
+
+def test_one_manual_host_tcp_success():
+    with SMB():
+        assert one_smb_service_discovered_message in runSMBeagleToCSVWithAuth("-h", "127.0.0.2")
+
+def test_one_manual_host_tcp_fail_if_not_listening():
+    with SMB("127.0.0.2"):
+        assert no_smb_service_discovered_message in runSMBeagleToCSVWithAuth("-h", "127.0.0.3")
+
+def test_two_manual_host_tcp_success():
+    with SMB("127.0.0.2"):
+        with SMB("127.0.0.3"):
+            assert two_smb_service_discovered_message in runSMBeagleToCSVWithAuth("-h", "127.0.0.2", "127.0.0.3")
+
+def test_one_manual_host_tcp_success_and_not_two_if_second_not_listening():
+    with SMB("127.0.0.2"):
+            assert one_smb_service_discovered_message in runSMBeagleToCSVWithAuth("-h", "127.0.0.2", "127.0.0.3")
+
+def test_one_discovered_host_tcp_success():
+    with SMB("127.0.0.2"):
+        assert one_smb_service_discovered_message in runSMBeagleToCSVWithAuth("-n", "127.0.0.0/24")
+
+def test_no_discovered_host_when_filtered():
+    with SMB("127.0.0.2"):
+        assert no_smb_service_discovered_message in runSMBeagleToCSVWithAuth("-n", "127.0.0.0/24","-H","127.0.0.2" )
+
+def test_one_discovered_host_when_one_filtered():
+    with SMB("127.0.0.2"):
+        with SMB("127.0.0.3"):
+            assert one_smb_service_discovered_message in runSMBeagleToCSVWithAuth("-n", "127.0.0.0/24","-H","127.0.0.2" )
+
+def test_two_discovered_host_tcp_success():
+    with SMB("127.0.0.2"):
+        with SMB("127.0.0.3"):
+            assert two_smb_service_discovered_message in runSMBeagleToCSVWithAuth("-n", "127.0.0.0/24")
+
+def test_three_discovered_host_tcp_success():
+    with SMB("127.0.0.2"):
+        with SMB("127.0.0.3"):
+            with SMB("127.0.0.4"):
+                assert three_smb_service_discovered_message in runSMBeagleToCSVWithAuth("-n", "127.0.0.0/24")
+
+def test_four_discovered_host_tcp_success():
+    with SMB("127.0.0.2"):
+        with SMB("127.0.0.3"):
+            with SMB("127.0.0.4"):
+                with SMB("127.0.0.5"):
+                    assert four_smb_service_discovered_message in runSMBeagleToCSVWithAuth("-n", "127.0.0.0/24")
+
+def test_disable_network_discovery():
+    no_networks_to_scan_message = "there are no networks or hosts to scan"
+    assert no_networks_to_scan_message in runSMBeagleToCSVWithAuth("-D")

tests/tests/test_030_output.py

@@ -0,0 +1,34 @@
+from helpers import *
+
+def test_no_acl_mode_returns_false_perms():
+    with SMB(dir_structure=["fileA","fileB","fileC"]):
+        for result in runSMBeagleToCSVWithAuthAndReturnResults("-h", "127.0.0.2", "-A"):
+            print(result)
+            # assert perms are all false
+            assert result["Readable"] == 'False'
+            assert result["Writeable"] == 'False'
+            assert result["Deletable"] == 'False'
+
+### test fast mode gives matching perms
+
+def test_csv_fields_exist():
+    with SMB(dir_structure=["fileA"]):
+        fields = ['Name','Host', 'Extension', 'Username', 'Hostname', 'UNCDirectory', 'CreationTime', 'LastWriteTime', 'Readable', 'Writeable', 'Deletable', 'DirectoryType', 'Base']
+        for result in runSMBeagleToCSVWithAuthAndReturnResults("-h", "127.0.0.2", "-A"):
+            for field in fields:
+                assert field in result.keys()
+
+def test_csv_fields_are_valid():
+    with SMB(dir_structure=[("dirA",["fileA.txt"])]):
+        fields = ['Name','Host', 'Extension', 'Username', 'Hostname', 'UNCDirectory', 'CreationTime', 'LastWriteTime', 'Readable', 'Writeable', 'Deletable', 'DirectoryType', 'Base']
+        for result in runSMBeagleToCSVWithAuthAndReturnResults("-h", "127.0.0.2"):
+            print(result)
+            assert result["Name"].lower() == "filea.txt"
+            assert result["Extension"].lower() == "txt"
+            assert result["Host"] == "127.0.0.2"
+            assert result["DirectoryType"] == "SMB"
+            assert result["UNCDirectory"].lower() ==  "\\\\127.0.0.2\\share\\dira"
+            assert result["Base"].lower() ==  "\\\\127.0.0.2\\share\\"
+            assert result["Readable"] == 'True'
+            assert result["Writeable"] == 'True'
+            assert result["Deletable"] == 'True'