From 512fcdc4e25dce8dfcb36704242a8aec5a10f4e2 Mon Sep 17 00:00:00 2001
From: Eric Bollengier <eric@baculasystems.com>
Date: Fri, 28 Jan 2022 16:56:49 +0100
Subject: [PATCH] Add command to list metadata owners for a given tenant

---
 bacula/src/cats/bdb.h       |  1 +
 bacula/src/cats/cats.c      | 18 +++++----
 bacula/src/cats/sql_list.c  | 77 +++++++++++++++++++++++++++++++++++--
 bacula/src/dird/ua_cmds.c   |  2 +-
 bacula/src/dird/ua_output.c | 16 +++++---
 5 files changed, 97 insertions(+), 17 deletions(-)

diff --git a/bacula/src/cats/bdb.h b/bacula/src/cats/bdb.h
index 65043aef8..8357f42a3 100644
--- a/bacula/src/cats/bdb.h
+++ b/bacula/src/cats/bdb.h
@@ -289,6 +289,7 @@ public:
    void bdb_list_snapshot_records(JCR *jcr, SNAPSHOT_DBR *sdbr,
               DB_LIST_HANDLER *sendit, void *ctx, e_list_type type);
    void bdb_list_files(JCR *jcr, FILE_DBR *fr, DB_RESULT_HANDLER *sendit, void *ctx);
+   void bdb_list_metadata_owner_records(JCR *jcr, META_DBR *meta_r, DB_LIST_HANDLER *sendit, void *ctx, e_list_type type);
    void bdb_list_metadata_records(JCR *jcr, META_DBR *meta_r, DB_LIST_HANDLER *sendit, void *ctx, e_list_type type);
    /* sql_update.c */
    bool bdb_update_job_start_record(JCR *jcr, JOB_DBR *jr);
diff --git a/bacula/src/cats/cats.c b/bacula/src/cats/cats.c
index fd8dcf57c..178262ce0 100644
--- a/bacula/src/cats/cats.c
+++ b/bacula/src/cats/cats.c
@@ -687,12 +687,6 @@ bool META_DBR::check()
       bsnprintf(errmsg, sizeof(errmsg), _("Type is not set"));
       return false;
    }
-
-   if (!Owner[0]) {
-      bsnprintf(errmsg, sizeof(errmsg), _("Owner is not set"));
-      return false;
-   }
-
    if (!Tenant[0]) {
       bsnprintf(errmsg, sizeof(errmsg), _("Tenant not set"));
       return false;
@@ -762,6 +756,12 @@ void META_DBR::create_db_filter(JCR *jcr, BDB *db, POOLMEM **where)
          pm_strcat(where, ") ");
       }
 
+      if (ClientName[0] != 0) {
+         db_escape_string(jcr, jcr->db, esc.c_str(), ClientName, strlen(ClientName));
+         Mmsg(tmp, " Client.Name='%s'", esc.c_str());
+         append_filter(where, tmp.c_str());
+      }
+
       if (ConversationId[0] != 0) {
          db_escape_string(jcr, jcr->db, esc.c_str(), ConversationId, strlen(ConversationId));
          Mmsg(tmp, " MetaEmail.EmailConversationId = '%s'", esc.c_str());
@@ -808,7 +808,11 @@ void META_DBR::create_db_filter(JCR *jcr, BDB *db, POOLMEM **where)
 
    if (Owner[0]) {
       db_escape_string(jcr, jcr->db, esc.c_str(), Owner, strlen(Owner));
-      Mmsg(tmp, " Meta%s.%sOwner = '%s'", Type, Type, esc.c_str());
+      if (strchr(Owner, '%')) {
+         Mmsg(tmp, " Meta%s.%sOwner ILIKE '%s'", Type, Type, esc.c_str());
+      } else {
+         Mmsg(tmp, " Meta%s.%sOwner = '%s'", Type, Type, esc.c_str());
+      }
       append_filter(where, tmp.c_str());
    }
 
diff --git a/bacula/src/cats/sql_list.c b/bacula/src/cats/sql_list.c
index 2c2173781..d0454b87a 100644
--- a/bacula/src/cats/sql_list.c
+++ b/bacula/src/cats/sql_list.c
@@ -1295,11 +1295,85 @@ void BDB::bdb_list_tag_records(JCR *jcr, TAG_DBR *tag, DB_LIST_HANDLER *result_h
    bdb_unlock();
 }
 
+/* List Owner in the metadata table */
+void BDB::bdb_list_metadata_owner_records(JCR *jcr, META_DBR *meta_r, DB_LIST_HANDLER *sendit, void *ctx, e_list_type type)
+{
+   POOL_MEM esc(PM_MESSAGE), tmp(PM_MESSAGE), where(PM_MESSAGE), join(PM_MESSAGE);
+
+   bdb_lock();
+   meta_r->create_db_filter(jcr, this, where.handle());
+   const char *where_filter = get_acls(DB_ACL_BIT(DB_ACL_JOB)     |
+                                       DB_ACL_BIT(DB_ACL_CLIENT), strcmp(where.c_str(), "") == 0);
+
+   const char *join_filter = (*where_filter && meta_r->ClientName[0] == 0) ?
+      get_acl_join_filter(DB_ACL_BIT(DB_ACL_CLIENT)) : "";
+
+   if (meta_r->ClientName[0] != 0) {
+      Mmsg(join, " JOIN Job ON (Job.JobId = Meta%s.JobId) JOIN Client USING (ClientId) ", meta_r->Type);
+
+   } else if (*where_filter) {  // We add manually the Job join filter part
+      Mmsg(join, " JOIN Job ON (Job.JobId = Meta%s.JobId) ", meta_r->Type);
+   }
+
+   if (where_filter && *where_filter) {
+      pm_strcat(where, where_filter);
+   }
+
+   if (join_filter && *join_filter) {
+      pm_strcat(join, join_filter);
+   }
+ 
+   if (meta_r->limit > 0) {
+      Mmsg(tmp, " LIMIT %d ", meta_r->limit);
+      pm_strcat(where, tmp.c_str());
+   }
+
+   if (meta_r->offset > 0) {
+      Mmsg(tmp, " OFFSET %ld ", meta_r->offset);
+      pm_strcat(where, tmp.c_str());
+   }
+
+   switch (type) {
+   case JSON_LIST:
+   case VERT_LIST:
+   case HORZ_LIST:
+      Mmsg(cmd,
+           "SELECT DISTINCT %sOwner "
+           "FROM Meta%s %s %s",
+           meta_r->Type, meta_r->Type, join.c_str(), where.c_str());
+         break;
+   default:
+         break;
+   }
+   Dmsg1(DT_SQL|50, "%s\n", cmd);
+   if (!QueryDB(jcr, cmd)) {
+      Jmsg(jcr, M_ERROR, 0, _("Query %s failed!\n"), cmd);
+      bdb_unlock();
+      return;
+   }
+   if (strcmp(meta_r->Type, "Email") == 0) {
+      Mmsg(esc, "metadataemail");
+   } else {
+      Mmsg(esc, "metaattachment");
+   }
+   
+   list_result(jcr, this, esc.c_str(), sendit, ctx, type);
+
+   sql_free_result();
+   bdb_unlock();
+}
+
 /*
  * List plugin objects (search is based on object provided)
  */
 void BDB::bdb_list_metadata_records(JCR *jcr, META_DBR *meta_r, DB_LIST_HANDLER *sendit, void *ctx, e_list_type type)
 {
+   if (!meta_r->Owner[0] || strchr(meta_r->Owner, '%')) {
+      /* List Owners */
+      bdb_list_metadata_owner_records(jcr, meta_r, sendit, ctx, type);
+      return;
+   }
+
    POOL_MEM esc(PM_MESSAGE), tmp(PM_MESSAGE), where(PM_MESSAGE), join(PM_MESSAGE);
    bdb_lock();
    //TODO add ACL part
@@ -1313,9 +1387,6 @@ void BDB::bdb_list_metadata_records(JCR *jcr, META_DBR *meta_r, DB_LIST_HANDLER
       get_acl_join_filter(DB_ACL_BIT(DB_ACL_CLIENT)) : "";
 
    if (meta_r->ClientName[0] != 0) {
-      bdb_escape_string(jcr, esc.c_str(), meta_r->ClientName, strlen(meta_r->ClientName));
-      Mmsg(tmp, " Client.Name='%s'", esc.c_str());
-      append_filter(where.handle(), tmp.c_str());
       Mmsg(join, " JOIN Job ON (Job.JobId = Meta%s.JobId) JOIN Client USING (ClientId) ", meta_r->Type);
 
    } else if (*where_filter) {  // We add manually the Job join filter part
diff --git a/bacula/src/dird/ua_cmds.c b/bacula/src/dird/ua_cmds.c
index 67e22d535..20c4a3ffb 100644
--- a/bacula/src/dird/ua_cmds.c
+++ b/bacula/src/dird/ua_cmds.c
@@ -138,7 +138,7 @@ static struct cmdstruct commands[] = {                                      /* C
        "\tevents [type=<str> | limit=<int> | order=<asc|desc> | days=<int> | start=<time-specification> | end=<time-specification> |\n"
        "\t\t source=<str> | code=<str> | type=<str> ]\n"
        "\tobjects [jobid=<jobid> client=<cli> type=<str> | category=<str> | status=<S> | limit=<int> | order=<asc|desc> ]\n"
-       "\tmetadata type=<email|attachment> tenant=<str> owner=<str>\n"
+       "\tmetadata type=<email|attachment> tenant=<str> [owner=<str>]\n"
        "\t    [jobid=<jobids> client=<cc> order=<Asc|desc> limit=<nn> orderby=<time> offset=<nn>]\n"
        "\t    [from=<str> to=<str> cc=<str> tags=<str> subject=<str> bodypreview=<str> all=<str>\n"
        "\t    minsize=<int> maxsize=<int> importance=<str> isread=<0|1> isdraft=<0|1> categories=<str>\n"
diff --git a/bacula/src/dird/ua_output.c b/bacula/src/dird/ua_output.c
index c071d4704..e9a238d9f 100644
--- a/bacula/src/dird/ua_output.c
+++ b/bacula/src/dird/ua_output.c
@@ -336,7 +336,8 @@ bail_out:
  *  list objects [type=objecttype job_id=id clientname=n,status=S] - list plugin objects
  *  list pluginrestoreconf jobid=x,y,z [id=k]
  *  list filemedia jobid=x fileindex=z
- *  list metadata type=[email|attachment] owner=xxx tenant=xxx jobid=<x,w,z> from=<str>
+ *  list metadata type=[email|attachment] tenant=xxx owner=xxx jobid=<x,w,z> client=<cli>
+ *             from=<str>
  *             to=<str> cc=<str> tags=<str> 
  *             subject=<str> bodypreview=<str> all=<str> minsize=<int> maxsize=<int> 
  *             importance=<str> isread=<0|1> isdraft=<0|1>
@@ -930,7 +931,10 @@ static int do_list_cmd(UAContext *ua, const char *cmd, e_list_type llist)
          META_DBR meta_r;
 
          for (j=i+1; j<ua->argc; j++) {
-            if (strcasecmp(ua->argk[j], NT_("jobid")) == 0 && ua->argv[j]) {
+            if (!ua->argv[j]) {
+               ua->error_msg(_("Invalid %s argument. Expecting value\n"), ua->argk[j]);
+               return 1;
+            } else if (strcasecmp(ua->argk[j], NT_("jobid")) == 0) {
                if (is_a_number_list(ua->argv[j]) && acl_access_jobid_ok(ua, ua->argv[j])) {
                   meta_r.JobIds = ua->argv[j];
                 } else {
@@ -1021,16 +1025,16 @@ static int do_list_cmd(UAContext *ua, const char *cmd, e_list_type llist)
             } else if (strcasecmp(ua->argk[j], NT_("hasattachment")) == 0) {
                meta_r.HasAttachment = str_to_uint64(ua->argv[j]);
 
-            } else if (strcasecmp(ua->argk[j], NT_("offset")) == 0 && ua->argv[j]) {
+            } else if (strcasecmp(ua->argk[j], NT_("offset")) == 0) {
                meta_r.offset = atoi(ua->argv[j]);
 
-            } else if (strcasecmp(ua->argk[j], NT_("limit")) == 0 && ua->argv[j]) {
+            } else if (strcasecmp(ua->argk[j], NT_("limit")) == 0) {
                meta_r.limit = atoi(ua->argv[j]);
 
-            } else if (strcasecmp(ua->argk[j], NT_("orderby")) == 0 && ua->argv[j]) {
+            } else if (strcasecmp(ua->argk[j], NT_("orderby")) == 0) {
                meta_r.orderby = bstrcasecmp(ua->argv[j], "Time") == 1;
 
-            } else if (strcasecmp(ua->argk[j], NT_("order")) == 0 && ua->argv[j]) {
+            } else if (strcasecmp(ua->argk[j], NT_("order")) == 0) {
                /* Other order are tested before */
                meta_r.order = bstrcasecmp(ua->argv[j], "DESC");
             }
-- 
2.47.3