ubuntu · GabrielNagy · Aug 9, 2023 · Aug 8, 2023 · Aug 8, 2023 · Aug 8, 2023
@@ -216,6 +216,7 @@ func TestPolicyUpdate(t *testing.T) {
 		readOnlyDirs        []string
 		winbindMockBehavior string
 		purge               bool
+		missingCertmonger   bool
 
 		wantErr bool
 	}{
@@ -603,6 +604,21 @@ func TestPolicyUpdate(t *testing.T) {
 			initState:    "localhost-uptodate",
 			systemAnswer: "no_proxy_object",
 		},
+		"Does not error when certmonger or cepces is not available": {
+			args:       []string{"-m"},
+			krb5ccname: "-",
+			krb5ccNamesState: []krb5ccNamesWithState{
+				{
+					src:     "ccache_EXAMPLE.COM",
+					machine: true,
+				},
+			},
+			initState: "localhost-uptodate",
+			addPaths: []string{
+				"lib/private", // make parent of private dir a file
+			},
+			missingCertmonger: true,
+		},
 
 		// Purge cases
 		"Purge current user policies": {
@@ -965,6 +981,17 @@ func TestPolicyUpdate(t *testing.T) {
 
 			t.Setenv("ADSYS_WBCLIENT_BEHAVIOR", tc.winbindMockBehavior)
 
+			// Create fake certmonger and cepces binaries for the certificate manager
+			if !tc.missingCertmonger {
+				binDir := t.TempDir()
+				for _, executable := range []string{"getcert", "cepces-submit"} {
+					// #nosec G306. We want this asset to be executable.
+					err := os.WriteFile(filepath.Join(binDir, executable), []byte("#!/bin/sh\necho $@\n"), 0755)
+					require.NoError(t, err, "Setup: could not create %q binary", executable)
+				}
+				t.Setenv("PATH", binDir+":"+os.Getenv("PATH"))
+			}
+
 			// Some tests will need some initial state assets
 			for _, k := range tc.clearDirs {
 				err := os.RemoveAll(filepath.Join(adsysDir, k))

@@ -0,0 +1 @@
+/usr/bin/baz {}
@@ -0,0 +1 @@
+/usr/bin/bar {}
@@ -0,0 +1 @@
+/usr/bin/foo {}
@@ -0,0 +1,6 @@
+^adsystestuser@example.com {
+/etc/environment r,
+@{HOMEDIRS}/.xauth* w,
+/usr/bin/{,b,d,rb}ash Ux,
+/usr/bin/{c,k,tc}sh Ux,
+}
@@ -0,0 +1,4 @@
+[org/gnome/desktop/interface]
+clock-format='24h'
+clock-show-date=false
+clock-show-weekday=true
@@ -0,0 +1,3 @@
+/org/gnome/desktop/interface/clock-format
+/org/gnome/desktop/interface/clock-show-date
+/org/gnome/desktop/interface/clock-show-weekday
@@ -0,0 +1 @@
+
@@ -0,0 +1 @@
+
@@ -0,0 +1,3 @@
+user-db:user
+system-db:gdm
+system-db:machine
@@ -0,0 +1 @@
+new content
@@ -0,0 +1 @@
+TDB file
@@ -0,0 +1,6 @@
+# This file is managed by adsys.
+# Do not edit this file manually.
+# Any changes will be overwritten.
+
+[Configuration]
+AdminIdentities=unix-user:bob@example.com;unix-group:mygroup@example2.com
@@ -0,0 +1 @@
+final machine script
@@ -0,0 +1 @@
+script user logon
@@ -0,0 +1 @@
+script user logoff
@@ -0,0 +1 @@
+script machine shutdown
@@ -0,0 +1 @@
+script machine startup
@@ -0,0 +1 @@
+script user logon
@@ -0,0 +1 @@
+subfolder other script
@@ -0,0 +1 @@
+unreferenced data
@@ -0,0 +1 @@
+unreferenced script
@@ -0,0 +1,2 @@
+scripts/script-machine-startup
+scripts/subfolder/other-script
@@ -0,0 +1,10 @@
+# This file is managed by adsys.
+# Do not edit this file manually.
+# Any changes will be overwritten.
+
+%admin	ALL=(ALL) !ALL
+%sudo	ALL=(ALL:ALL) !ALL
+
+"bob@example.com"	ALL=(ALL:ALL) ALL
+"%mygroup@example2.com"	ALL=(ALL:ALL) ALL
+
@@ -13,6 +13,7 @@ from samba.dcerpc import preg
 
 from vendor_samba.gp.gpclass import GPOStorage
 from vendor_samba.gp import gp_cert_auto_enroll_ext as cae
+from vendor_samba.gp.util.logging import logger_init, log
 
 class adsys_cert_auto_enroll(cae.gp_cert_auto_enroll_ext):
     def enroll(self, guid, entries, trust_dir, private_dir, global_trust_dir):
@@ -57,17 +58,24 @@ def main():
     private_dir = os.path.join(args.state_dir, 'private', 'certs')
     global_trust_dir = args.global_trust_dir
 
-    # Create needed directories if they don't exist
-    for directory in [samba_cache_dir, trust_dir, private_dir, global_trust_dir]:
-        if not os.path.exists(directory):
-            perms = 0o700 if directory == private_dir else 0o755
-            os.makedirs(directory, mode=perms)
-
     with tempfile.NamedTemporaryFile(prefix='smb_conf') as smb_conf:
         smb_conf.write(smb_config(args.realm, args.debug).encode('utf-8'))
         smb_conf.flush()
 
         lp = param.LoadParm(smb_conf.name)
+        # Set up logging
+        logger_init('cert-autoenroll', lp.log_level())
+
+        if not cepces_submit() or not certmonger():
+            log.warning('certmonger and/or cepces not found, skipping certificate enrollment')
+            return
+
+        # Create needed directories if they don't exist
+        for directory in [samba_cache_dir, trust_dir, private_dir, global_trust_dir]:
+            if not os.path.exists(directory):
+                perms = 0o700 if directory == private_dir else 0o755
+                os.makedirs(directory, mode=perms)
+
         c = Credentials()
         c.set_kerberos_state(MUST_USE_KERBEROS)
         c.guess(lp)
@@ -122,6 +130,13 @@ def gpo_entries(entries_json):
             raise ValueError(f'GPO data must be a JSON array of objects') from exc
     return entries
 
+def cepces_submit():
+    certmonger_dirs = [os.environ.get('PATH'), '/usr/lib/certmonger',
+                       '/usr/libexec/certmonger']
+    return shutil.which('cepces-submit', path=':'.join(certmonger_dirs))
+
+def certmonger():
+    return shutil.which('getcert')
 
 if __name__ == "__main__":
     sys.exit(main())
@@ -80,6 +80,9 @@ func TestCertAutoenrollScript(t *testing.T) {
 		readOnlyPath    bool
 		autoenrollError bool
 
+		missingCertmonger bool
+		missingCepces     bool
+
 		wantErr bool
 	}{
 		"Enroll with simple configuration":                   {args: []string{"enroll", "keypress", "example.com"}},
@@ -89,6 +92,10 @@ func TestCertAutoenrollScript(t *testing.T) {
 
 		"Unenroll": {args: []string{"unenroll", "keypress", "example.com"}},
 
+		// Missing binary cases
+		"Enroll with certmonger not installed": {args: []string{"enroll", "keypress", "example.com"}, missingCertmonger: true},
+		"Enroll with cepces not installed":     {args: []string{"enroll", "keypress", "example.com"}, missingCepces: true},
+
 		// Error cases
 		"Error on missing arguments": {args: []string{"enroll"}, wantErr: true},
 		"Error on invalid flags":     {args: []string{"enroll", "keypress", "example.com", "--invalid_flag"}, wantErr: true},
@@ -109,6 +116,17 @@ func TestCertAutoenrollScript(t *testing.T) {
 			stateDir := t.TempDir()
 			sambaCacheDir := filepath.Join(stateDir, "samba")
 			globalTrustDir := filepath.Join(stateDir, "ca-certificates")
+			binDir := t.TempDir()
+			if !tc.missingCertmonger {
+				// #nosec G306. We want this asset to be executable.
+				err := os.WriteFile(filepath.Join(binDir, "getcert"), []byte("#!/bin/sh\necho $@\n"), 0755)
+				require.NoError(t, err, "Setup: could not create getcert binary")
+			}
+			if !tc.missingCepces {
+				// #nosec G306. We want this asset to be executable.
+				err := os.WriteFile(filepath.Join(binDir, "cepces-submit"), []byte("#!/bin/sh\necho $@\n"), 0755)
+				require.NoError(t, err, "Setup: could not create cepces binary")
+			}
 
 			// Create a dummy cache file to ensure we don't fail when removing a non-empty directory
 			testutils.CreatePath(t, filepath.Join(sambaCacheDir, "cert_gpo_state_HOST.tdb"))
@@ -121,7 +139,10 @@ func TestCertAutoenrollScript(t *testing.T) {
 
 			// #nosec G204: we control the command line name and only change it for tests
 			cmd := exec.Command(certAutoenrollCmd, args...)
-			cmd.Env = append(os.Environ(), "PYTHONPATH="+pythonPath)
+			cmd.Env = append(os.Environ(),
+				"PYTHONPATH="+pythonPath,
+				"PATH="+binDir+":"+os.Getenv("PATH"),
+			)
 			if tc.autoenrollError {
 				cmd.Env = append(os.Environ(), "ADSYS_WANT_AUTOENROLL_ERROR=1")
 			}

@@ -16,7 +16,6 @@
 package certificate
 
 import (
-	"bytes"
 	"context"
 	_ "embed" // embed cert enroll python script
 	"encoding/json"
@@ -247,16 +246,14 @@ func (m *Manager) runScript(ctx context.Context, action, objectName string, extr
 		fmt.Sprintf("KRB5CCNAME=%s", filepath.Join(m.krb5CacheDir, objectName)),
 		fmt.Sprintf("PYTHONPATH=%s:%s", os.Getenv("PYTHONPATH"), m.vendorPythonDir),
 	)
-	var stdout, stderr bytes.Buffer
-	cmd.Stdout = &stdout
-	cmd.Stderr = &stderr
 	smbsafe.WaitExec()
 	defer smbsafe.DoneExec()
 
-	if err := cmd.Run(); err != nil {
-		return fmt.Errorf(i18n.G("failed to run certificate autoenrollment script (exited with %d): %v\n%s"), cmd.ProcessState.ExitCode(), err, stderr.String())
+	output, err := cmd.CombinedOutput()
+	if err != nil {
+		return fmt.Errorf(i18n.G("failed to run certificate autoenrollment script (exited with %d): %v\n%s"), cmd.ProcessState.ExitCode(), err, string(output))
 	}
-	log.Infof(ctx, i18n.G("Certificate autoenrollment script ran successfully\n%s"), stdout.String())
+	log.Infof(ctx, i18n.G("Certificate autoenrollment script ran successfully\n%s"), string(output))
 	return nil
 }
 

@@ -0,0 +1,5 @@
+Loading smb.conf
+[global]
+realm = example.com
+
+WARNING: certmonger and/or cepces not found, skipping certificate enrollment
@@ -0,0 +1,5 @@
+Loading smb.conf
+[global]
+realm = example.com
+
+WARNING: certmonger and/or cepces not found, skipping certificate enrollment
@@ -1,6 +1,10 @@
-def LoadParm(smb_conf=None):
-    if smb_conf is None:
-        return
-    print('Loading smb.conf')
-    with open(smb_conf, 'r') as f:
-        print(f.read())
+class LoadParm(object):
+    def __init__(self, smb_conf=None):
+        if smb_conf is None:
+            return
+        print('Loading smb.conf')
+        with open(smb_conf, 'r') as f:
+            print(f.read())
+
+    def log_level(self):
+        return 0
@@ -0,0 +1,7 @@
+def logger_init(_name, _level):
+    pass
+
+class log(object):
+    @staticmethod
+    def warning(msg):
+        print(f'WARNING: {msg}')