summaryrefslogtreecommitdiffstats
path: root/nixos/tests/openbao.nix
blob: e5510f5bf2c1efb33544896ac69d10f3d3fff613 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
{ lib, ... }:
let
  certs = import ./common/acme/server/snakeoil-certs.nix;
  domain = certs.domain;
in
{
  name = "openbao";

  meta.maintainers = with lib.maintainers; [ kranzes ];

  nodes.machine =
    { config, ... }:
    {
      security.pki.certificateFiles = [ certs.ca.cert ];

      networking.extraHosts = ''
        127.0.0.1 ${domain}
      '';

      services.openbao = {
        enable = true;

        settings = {
          ui = true;

          listener = {
            default = {
              type = "tcp";
              tls_cert_file = certs.${domain}.cert;
              tls_key_file = certs.${domain}.key;
            };

            unix = {
              type = "unix";
            };

            unix-custom = {
              type = "unix";
              address = "/run/openbao/world-accessible.sock";
              socket_mode = "0222";
              socket_user = "openbao";
              socket_group = "openbao";
            };
          };

          cluster_addr = "https://127.0.0.1:8201";
          api_addr = "https://${domain}:8200";

          storage.raft.path = "/var/lib/openbao";
        };
      };

      environment.variables = {
        BAO_ADDR = config.services.openbao.settings.api_addr;
        BAO_FORMAT = "json";
      };
    };

  testScript =
    { nodes, ... }:
    let
      inherit (nodes.machine.services.openbao.settings) listener;
    in
    ''
      import json

      start_all()

      with subtest("Wait for OpenBao to start up"):
        machine.wait_for_unit("openbao.service")
        machine.wait_for_open_port(8200)
        machine.wait_for_open_unix_socket("${listener.unix.address}")
        machine.wait_for_open_unix_socket("${listener.unix-custom.address}")

      with subtest("Check Unix Socket listeners"):
        t.assertEqual("srwx------ openbao:openbao", machine.succeed("stat --printf '%A %U:%G' ${listener.unix.address}"))
        machine.fail("su -l nobody -s /bin/sh -c 'curl --fail --silent --unix-socket ${listener.unix.address} localhost'")

        t.assertEqual("s-w--w--w- openbao:openbao", machine.succeed("stat --printf '%A %U:%G' ${listener.unix-custom.address}"))
        machine.succeed("su -l nobody -s /bin/sh -c 'curl --fail --silent --unix-socket ${listener.unix-custom.address} localhost'")

      with subtest("Check that the web UI is being served"):
        machine.succeed("curl -L --fail --show-error --silent $BAO_ADDR | grep '<title>OpenBao</title>'")

      with subtest("Check that OpenBao is not initialized"):
        status_output = json.loads(machine.fail("bao status"))
        assert not status_output["initialized"]

      with subtest("Initialize OpenBao"):
        init_output = json.loads(machine.succeed("bao operator init"))

      with subtest("Check that OpenBao is initialized and sealed"):
        status_output = json.loads(machine.fail("bao status"))
        assert status_output["initialized"]
        assert status_output["sealed"]

      with subtest("Unseal OpenBao"):
        for key in init_output["unseal_keys_b64"][:init_output["unseal_threshold"]]:
          machine.succeed(f"bao operator unseal {key}")

      with subtest("Check that OpenBao is not sealed"):
        status_output = json.loads(machine.succeed("bao status"))
        assert not status_output["sealed"]

      with subtest("Login with root token"):
        machine.succeed(f"bao login {init_output["root_token"]}")

      with subtest("Enable userpass auth method"):
        machine.succeed("bao auth enable userpass")

      with subtest("Create a user in userpass"):
        machine.succeed("bao write auth/userpass/users/testuser password=testpassword")

      with subtest("Login to a user from userpass"):
        machine.succeed("bao login -method userpass username=testuser password=testpassword")

      with subtest("Write a secret to cubbyhole"):
        machine.succeed("bao write cubbyhole/my-secret my-value=s3cr3t")

      with subtest("Read a secret from cubbyhole"):
        read_output = json.loads(machine.succeed("bao read cubbyhole/my-secret"))
        assert read_output["data"]["my-value"] == "s3cr3t"
    '';
}