mailnix/tests/rspamd.nix

{pkgs, ...}:
with (import ./common/lib.nix {inherit pkgs;}); let
  lib = pkgs.lib;
  accounts = {
    "normal" = {
      address = "user@example.com";
      password = "secret-password1";
    };
    "normal2" = {
      address = "user@example.org";
      password = "secret-password2;";
    };
  };
  genDkimSecret = domain: name: type:
    pkgs.runCommand "mk-dkim-secrets-${domain}-${selector}" {
      buildInputs = [pkgs.rspamd];
      inherit domain name type;
    } ''
      rspamadm dkim_keygen -d $domain -s $name -t $type ${lib.optionalString (type == "rsa") "-b 2048"} -k $out
    '';

  mkDkimSettings = domains: selectors:
    lib.listToAttrs (
      map (domain:
        lib.nameValuePair domain (map (entry: {
            selector = entry.name;
            keyFile = genDkimSecret domain entry.name entry.type;
          })
          selectors))
      domains
    );
in
  pkgs.nixosTest {
    name = "rspamd";
    nodes = {
      server = {pkgs, ...}: {
        imports = [./common/server.nix];
        mailsystem = {
          fqdn = "mail.example.com";
          domains = ["example.com" "example.org"];
          accounts = mkAccounts accounts;
          dkimSettings = mkDkimSettings ["example.com" "example.org"] [
            {
              name = "elliptic";
              type = "ed25519";
            }
            {
              name = "selector";
              type = "rsa";
            }
          ];
        };
      };
      client = {...}: {
        imports = [./common/client.nix];
      };
    };
    testScript = {nodes, ...}: let
      cfg = nodes.server.mailsystem;
      serverAddr = nodes.server.networking.primaryIPAddress;
      clientAddr = nodes.client.networking.primaryIPAddress;
      smtpSettings = {
        address = serverAddr;
        port = 465;
      };
      sendMail = mkSendMail smtpSettings accounts;
      recvMail = mkRecvMail serverAddr accounts;
      test-mark-spam = accountName:
        pkgs.writeScript "imap-mark-spam" ''
          #!${pkgs.python3.interpreter}
          import imaplib

          with imaplib.IMAP4_SSL('${serverAddr}') as imap:
            imap.login('${accounts."${accountName}".address}', '${accounts."${accountName}".password}')
            imap.select()
            status, [response] = imap.search(None, 'ALL')
            msg_ids = response.decode("utf-8").split(' ')
            print(msg_ids)
            assert status == 'OK'
            assert len(msg_ids) == 1

            imap.copy(','.join(msg_ids), 'Junk')
            for num in msg_ids:
              imap.store(num, '+FLAGS', '\\Deleted')
            imap.expunge()

            imap.select('Junk')
            status, [response] = imap.search(None, 'ALL')
            msg_ids = response.decode("utf-8").split(' ')
            print(msg_ids)
            assert status == 'OK'
            assert len(msg_ids) == 1

            imap.close()
        '';
      test-mark-ham = accountName:
        pkgs.writeScript "imap-mark-ham" ''
          #!${pkgs.python3.interpreter}
          import imaplib

          with imaplib.IMAP4_SSL('${serverAddr}') as imap:
            imap.login('${accounts."${accountName}".address}', '${accounts."${accountName}".password}')
            imap.select('Junk')
            status, [response] = imap.search(None, 'ALL')
            msg_ids = response.decode("utf-8").split(' ')
            print(msg_ids)
            assert status == 'OK'
            assert len(msg_ids) == 1

            imap.copy(','.join(msg_ids), 'INBOX')
            for num in msg_ids:
              imap.store(num, '+FLAGS', '\\Deleted')
            imap.expunge()

            imap.select('INBOX')
            status, [response] = imap.search(None, 'ALL')
            msg_ids = response.decode("utf-8").split(' ')
            print(msg_ids)
            assert status == 'OK'
            assert len(msg_ids) == 1

            imap.close()
        '';
    in ''
      start_all()

      server.wait_for_unit("multi-user.target")
      client.wait_for_unit("multi-user.target")
      server.wait_until_succeeds("${waitForRspamd nodes.server}")

      with subtest("rspamd configuration is valid"):
          server.succeed("${pkgs.rspamd}/bin/rspamadm configtest >&2")

      with subtest("rspamd rejects spam"):
          client.fail("${sendMail "normal" "" accounts."normal2".address ''
        Subject: GTUBE-Test

        Hello User2,
        this is a mail containing a GTUBE pattern that should result in the rejection of this mail.

        XJS*C4JDBQADN1.NSBN3*2IDNEN*GTUBE-STANDARD-ANTI-UBE-TEST-EMAIL*C.34X
      ''}")

      with subtest("imap sieve junk trainer"):
          client.succeed("${sendMail "normal" "" accounts."normal2".address ''
        Subject: Testmail

        Hello User2,
        this is a testmail.
      ''}")
          server.wait_until_fails('${pendingPostqueue}')

          client.succeed("${test-mark-spam "normal2"} >&2")
          server.wait_until_succeeds("journalctl -u dovecot2 | grep -i learn-spam.sh >&2")
          server.fail("journalctl -u dovecot2 | grep -i learn-spam.sh | grep -i error >&2")

          client.succeed("${test-mark-ham "normal2"} >&2")
          server.wait_until_succeeds("journalctl -u dovecot2 | grep -i learn-ham.sh >&2")
          server.fail("journalctl -u dovecot2 | grep -i learn-ham.sh | grep -i error >&2")

      with subtest("dkim signing"):
          client.succeed("${sendMail "normal2" "" accounts."normal".address ''
        Subject: Testmail

        Hello User1,
        this is also a testmail.
      ''}")
          server.wait_until_fails('${pendingPostqueue}')
          client.execute("${cleanupMail}")
          # fetchmail returns EXIT_CODE 0 when it retrieves mail
          client.succeed("${recvMail "normal"} >&2")

          client.succeed("cat ~/mail/* >&2")
          # make sure the mail has all configured dkim signatures
          client.succeed("grep ${(builtins.elemAt cfg.dkimSettings."example.com" 0).selector} ~/mail/*")
          client.succeed("grep ${(builtins.elemAt cfg.dkimSettings."example.com" 1).selector} ~/mail/*")
    '';
  }