Resolves: bug 249366

Bug Description: rhds71 - search filters returns too many entries on integer attributes value greater than 2 to the power of 31
Reviewed by: nkinder, nhosoi (Thanks!)
Fix Description: The way >= and <= searches are supposed to work in LDAP is that you are supposed to define an ORDERING matching rule for the attribute you want to use in the search filter.  The way our code is written, most strings "just work" as a side effect of the way bdb sorts the keys by default - so you can do (uid>=jvedder) and get what you would expect, even though LDAP says this is illegal because the schema definition of the uid attribute does not have an ORDERING matching rule.  And INTEGER worked with the old binary format for the same reason.  The only attribute definitions we use with ORDERING are attributes that use Generalized Time syntax (e.g. createTimestamp, et. al.) and numSubordinates (which uses INTEGER, but this is a special case handled internally by the db code).
The way it works now is that the indexing code will honor the ORDERING matching rule specified in the schema definition.  Or, if ORDERING is not specified, the user can use the nsMatchingRule index configuration.  This will allow an existing customer that depends all integer syntax attributes (e.g. uidNumber) to allow range searches by default to enable range searches without editing the schema.  The syntax definition for the attribute must also specify a compare function.  This compare function will be used by the bdb bt_compare() function.
I also fixed a bug in the integer normalize code - a string of all zeros should normalize to a single "0".  In all other cases, the leading zeros should be removed.
Platforms tested: RHEL5 x86_64
Flag Day: Yes.  Integer indexes will need to be rebuilt (except for numsubordinates).
Doc impact: Yes - document slapi API additions
QA impact: Pay close attention to tests that use >= or <= search filters, both with and without index attributes.  Also, pay close attention to greater/less than searches using i18n collations.
New Tests integrated into TET: Forthcoming

ldap/servers/plugins/syntaxes/cis.c

@@ -84,6 +84,19 @@ static char *boolean_names[] = { "Boolean", BOOLEAN_SYNTAX_OID, 0 };
 static char *time_names[] = { "GeneralizedTime", "time",
 		GENERALIZEDTIME_SYNTAX_OID, 0 };
 
+#define GENERALIZEDTIMEMATCH_OID "2.5.13.27"
+#define GENERALIZEDTIMEORDERINGMATCH_OID "2.5.13.28"
+static Slapi_MatchingRuleEntry
+generalizedTimeMatch = { GENERALIZEDTIMEMATCH_OID, NULL /* no alias? */,
+                         "generalizedTimeMatch", "The rule evaluates to TRUE if and only if the attribute value represents the same universal coordinated time as the assertion value.",
+                         GENERALIZEDTIME_SYNTAX_OID, 0 /* not obsolete */ };
+
+static Slapi_MatchingRuleEntry
+generalizedTimeOrderingMatch = { GENERALIZEDTIMEORDERINGMATCH_OID, NULL /* no alias? */,
+                                 "generalizedTimeOrderingMatch", "The rule evaluates to TRUE if and only if the attribute value represents a universal coordinated time that is earlier than the universal coordinated time represented by the assertion value.",
+                                 GENERALIZEDTIME_SYNTAX_OID, 0 /* not obsolete */ };
+
+
 static char *country_names[] = { "Country String",
 		COUNTRYSTRING_SYNTAX_OID, 0};
 
@@ -223,6 +236,9 @@ time_init( Slapi_PBlock *pb )
 	LDAPDebug( LDAP_DEBUG_PLUGIN, "=> time_init\n", 0, 0, 0 );
 	rc = register_cis_like_plugin( pb, &time_pdesc, time_names,
 			GENERALIZEDTIME_SYNTAX_OID );
+	/* also register this plugin for matching rules */
+	rc |= slapi_matchingrule_register(&generalizedTimeMatch);
+	rc |= slapi_matchingrule_register(&generalizedTimeOrderingMatch);
 	LDAPDebug( LDAP_DEBUG_PLUGIN, "<= time_init %d\n", rc, 0, 0 );
 	return( rc );
 }

ldap/servers/plugins/syntaxes/int.c

@@ -58,9 +58,23 @@ static int int_compare(struct berval	*v1, struct berval	*v2);
 /* the first name is the official one from RFC 2252 */
 static char *names[] = { "INTEGER", "int", INTEGER_SYNTAX_OID, 0 };
 
+#define INTEGERMATCH_OID "2.5.13.14"
+#define INTEGERORDERINGMATCH_OID "2.5.13.15"
+
 static Slapi_PluginDesc pdesc = { "int-syntax", PLUGIN_MAGIC_VENDOR_STR,
 	PRODUCTTEXT, "integer attribute syntax plugin" };
 
+static Slapi_MatchingRuleEntry
+integerMatch = { INTEGERMATCH_OID, NULL /* no alias? */,
+                 "integerMatch", "The rule evaluates to TRUE if and only if the attribute value and the assertion value are the same integer value.",
+                 INTEGER_SYNTAX_OID, 0 /* not obsolete */ };
+
+static Slapi_MatchingRuleEntry
+integerOrderingMatch = { INTEGERORDERINGMATCH_OID, NULL /* no alias? */,
+                         "integerOrderingMatch", "The rule evaluates to TRUE if and only if the integer value of the attribute value is less than the integer value of the assertion value.",
+                         INTEGER_SYNTAX_OID, 0 /* not obsolete */ };
+
+
 int
 int_init( Slapi_PBlock *pb )
 {
@@ -88,6 +102,10 @@ int_init( Slapi_PBlock *pb )
 	rc |= slapi_pblock_set( pb, SLAPI_PLUGIN_SYNTAX_COMPARE,
 	    (void *) int_compare );
 
+	/* also register this plugin for matching rules */
+	rc |= slapi_matchingrule_register(&integerMatch);
+	rc |= slapi_matchingrule_register(&integerOrderingMatch);
+
 	LDAPDebug( LDAP_DEBUG_PLUGIN, "<= int_init %d\n", rc, 0, 0 );
 	return( rc );
 }

ldap/servers/plugins/syntaxes/value.c

@@ -117,21 +117,25 @@ value_normalize(
 	/* have to do this after trimming spaces */
 	if (syntax & SYNTAX_INT) {
 		int foundsign = 0;
+		int foundzero = 0;
+
 		if (*s == '-') {
 			foundsign = 1;
 			LDAP_UTF8INC(s);
 		}
 
 		while (*s && (*s == '0')) {
+			foundzero = 1;
 			LDAP_UTF8INC(s);
 		}
 
-		/* if there is a hyphen, make sure it is just to the left
-		   of the first significant (i.e. non-zero) digit e.g.
-		   convert -00000001 to -1 */
-		if (foundsign && (s > d)) {
-			*d = '-';
-			d++;
+		if (foundzero && !*s) { /* value is all zeros */
+			*d++ = '0'; /* set value to a single zero */
+		} else if (foundsign && (s > d)) {
+			/* if there is a hyphen, make sure it is just to the left
+			   of the first significant (i.e. non-zero) digit e.g.
+			   convert -00000001 to -1 */
+			*d++ = '-';
 		}
 		/* s should now point at the first significant digit/char */
 	}

ldap/servers/slapd/back-ldbm/back-ldbm.h

@@ -364,12 +364,22 @@ struct attrinfo {
                                          * yet. */
 
 #define	IS_INDEXED( a )	( a & INDEX_ANY )
-	void	*ai_plugin;
+	void	*ai_plugin; /* the syntax plugin for this attribute */
 	char	**ai_index_rules; /* matching rule OIDs */
 	void	*ai_dblayer;	  /* private data used by the dblayer code */
 	PRInt32 ai_dblayer_count; /* used by the dblayer code */
 	idl_private	*ai_idl;  /* private data used by the IDL code (eg locking the IDLs) */
 	attrcrypt_private	*ai_attrcrypt;  /* private data used by the attribute encryption code (eg is it enabled or not) */
+	value_compare_fn_type ai_key_cmp_fn; /* function used to compare two index keys -
+											The function is the compare function provided by
+											ai_plugin - this function is used to order
+											the keys in the index so that we can use ORDERING
+											searches.  In order for this function to be used,
+											the syntax plugin must define a compare function,
+											and either the attribute definition schema must
+											specify an ORDERING matching rule, or the index
+											configuration must define an ORDERING matching rule.
+										 */
 };
 
 #define MAXDBCACHE	20

ldap/servers/slapd/back-ldbm/dblayer.c

@@ -223,6 +223,33 @@ static int dblayer_db_remove_ex(dblayer_private_env *env, char const path[], cha
 #define MEGABYTE (1024 * 1024)
 #define GIGABYTE (1024 * MEGABYTE)
 
+
+/* This function compares two index keys.  It is assumed
+   that the values are already normalized, since they should have
+   been when the index was created (by int_values2keys).
+
+   richm - actually, the current syntax compare functions
+   always normalize both arguments.  We need to add an additional
+   syntax compare function that does not normalize or takes
+   an argument like value_cmp to specify to normalize or not.
+*/
+
+typedef int (*syntax_cmp_fn_type)(struct berval *, struct berval *);
+static int
+dblayer_bt_compare(DB *db, const DBT *dbt1, const DBT *dbt2)
+{
+    struct berval bv1, bv2;
+    value_compare_fn_type syntax_cmp_fn = (value_compare_fn_type)db->app_private;
+
+    bv1.bv_val = (char *)dbt1->data+1; /* remove leading '=' */
+    bv1.bv_len = (ber_len_t)dbt1->size-1;
+
+    bv2.bv_val = (char *)dbt2->data+1; /* remove leading '=' */
+    bv2.bv_len = (ber_len_t)dbt2->size-1;
+
+    return syntax_cmp_fn(&bv1, &bv2);
+}
+
 /* this flag use if user remotely turned batching off */
 
 #define FLUSH_REMOTEOFF -1 
@@ -2594,7 +2621,7 @@ int    dblayer_flush(struct ldbminfo *li)
        Success: 0
     Failure: -1
  */
-int dblayer_open_file(backend *be, char* indexname, int open_flag, int index_flags, DB **ppDB)
+int dblayer_open_file(backend *be, char* indexname, int open_flag, struct attrinfo *ai, DB **ppDB)
 {
     struct ldbminfo *li = (struct ldbminfo *) be->be_database->plg_private;
     ldbm_instance *inst = (ldbm_instance *) be->be_instance_info;
@@ -2682,7 +2709,7 @@ int dblayer_open_file(backend *be, char* indexname, int open_flag, int index_fla
     }
 #endif
 
-    if (idl_get_idl_new() && !(index_flags & INDEX_VLV)) {
+    if (idl_get_idl_new() && !(ai->ai_indexmask & INDEX_VLV)) {
         return_value = dbp->set_flags(dbp, DB_DUP | DB_DUPSORT);
         if (0 != return_value)
             goto out;
@@ -2695,7 +2722,7 @@ int dblayer_open_file(backend *be, char* indexname, int open_flag, int index_fla
             goto out;
     }
 
-    if (index_flags & INDEX_VLV) {
+    if (ai->ai_indexmask & INDEX_VLV) {
         /*
          * Need index with record numbers for
          * Virtual List View index
@@ -2703,6 +2730,24 @@ int dblayer_open_file(backend *be, char* indexname, int open_flag, int index_fla
         return_value = dbp->set_flags(dbp, DB_RECNUM);
         if (0 != return_value)
             goto out;
+    } else if (ai->ai_key_cmp_fn) { /* set in attr_index_config() */
+        /*
+          This is so that we can have ordered keys in the index, so that
+          greater than/less than searches work on indexed attrs.  We had
+          to introduce this when we changed the integer key format from
+          a 32/64 bit value to a normalized string value.  The default
+          bdb key cmp is based on length and lexicographic order, which
+          does not work with integer strings.
+
+          NOTE: If we ever need to use app_private for something else, we
+          will have to create some sort of data structure with different
+          fields for different uses.  We will also need to have a new()
+          function that creates and allocates that structure, and a
+          destroy() function that destroys the structure, and make sure
+          to call it when the DB* is closed and/or freed.
+        */
+        dbp->app_private = (void *)ai->ai_key_cmp_fn;
+        dbp->set_bt_compare(dbp, dblayer_bt_compare);
     }
 
     /* The subname argument allows applications to have
@@ -2842,7 +2887,7 @@ int dblayer_get_index_file(backend *be, struct attrinfo *a, DB** ppDB, int open_
    * index file and stuff it in the attrinfo.
    */
   return_value = dblayer_open_file(be, attribute_name, open_flags,
-                                   a->ai_indexmask, &pDB);
+                                   a, &pDB);
   if (0 == return_value) {
       /* Opened it OK */
       dblayer_handle *handle = (dblayer_handle *) 

ldap/servers/slapd/back-ldbm/import-merge.c

@@ -338,7 +338,7 @@ static int import_count_merge_input_files(ldbm_instance *inst,
     return 0;
 }
 
-static int import_open_merge_input_files(backend *be, char *indexname,
+static int import_open_merge_input_files(backend *be, IndexInfo *index_info,
 	int passes, DB ***input_files, int *number_found, int *pass_number)
 {
     int i = 0;
@@ -354,16 +354,22 @@ static int import_open_merge_input_files(backend *be, char *indexname,
     }
     for (i = 0; i < passes; i++) {
 	DB *pDB = NULL;
-	char *filename = slapi_ch_smprintf("%s.%d", indexname, i+1);
+	char *filename = slapi_ch_smprintf("%s.%d", index_info->name, i+1);
 
 	if (NULL == filename) {
 	    return -1;
 	}
 
 	if (vlv_isvlv(filename)) {
-		ret = dblayer_open_file(be, filename, 0, INDEX_VLV, &pDB);
+		/* not sure why the file would be marked as a vlv index but
+		   not the index configuration . . . but better make sure
+		   the new code works with the old semantics */
+		int saved_mask = index_info->ai->ai_indexmask;
+		index_info->ai->ai_indexmask |= INDEX_VLV;
+		ret = dblayer_open_file(be, filename, 0, index_info->ai, &pDB);
+		index_info->ai->ai_indexmask = saved_mask;
 	} else {
-		ret = dblayer_open_file(be, filename, 0, 0, &pDB);
+		ret = dblayer_open_file(be, filename, 0, index_info->ai, &pDB);
 	}
 
 	slapi_ch_free( (void**)&filename);
@@ -488,7 +494,7 @@ static int import_merge_one_file(ImportWorkerInfo *worker, int passes,
         }
 #endif
 
-	ret = import_open_merge_input_files(be, worker->index_info->name,
+	ret = import_open_merge_input_files(be, worker->index_info,
 		passes, &input_files, &number_found, &pass_number);
 	if (0 != ret) {
 	    import_log_notice(worker->job, "MERGE FAIL 10");
@@ -496,7 +502,7 @@ static int import_merge_one_file(ImportWorkerInfo *worker, int passes,
 	}
 
 	ret = dblayer_open_file(be, worker->index_info->name, 1,
-				vlv_index ? INDEX_VLV : 0, &output_file);
+				worker->index_info->ai, &output_file);
 	if (0 != ret) {
 	    import_log_notice(worker->job, "Failed to open output file for "
 			      "index %s in merge", worker->index_info->name);

ldap/servers/slapd/back-ldbm/ldbm_attr.c

@@ -57,6 +57,7 @@ attrinfo_new()
 	p->ai_dblayer= NULL;
         p->ai_dblayer_count = 0;
 	p->ai_idl= NULL;
+	p->ai_key_cmp_fn = NULL;
     return p;
 }
 
@@ -66,6 +67,7 @@ attrinfo_delete(struct attrinfo **pp)
     if(pp!=NULL && *pp!=NULL)
     {
         idl_release_private(*pp);
+        (*pp)->ai_key_cmp_fn = NULL;
         slapi_ch_free((void**)&((*pp)->ai_type));
         slapi_ch_free((void**)(*pp)->ai_index_rules);
         slapi_ch_free((void**)pp);
@@ -179,9 +181,12 @@ attr_index_config(
 		}
 	}
 	for ( i = 0; attrs[i] != NULL; i++ ) {
+		int need_compare_fn = 0;
+		char *attrsyntax_oid = NULL;
 		a = attrinfo_new();
 		a->ai_type = slapi_attr_basetype( attrs[i], NULL, 0 );
 		slapi_attr_type2plugin( a->ai_type, &a->ai_plugin );
+		attrsyntax_oid = slapi_ch_strdup(plugin_syntax2oid(a->ai_plugin));
 		if ( argc == 1 ) {
 			a->ai_indexmask = (INDEX_PRESENCE | INDEX_EQUALITY |
 			    INDEX_APPROX | INDEX_SUB);
@@ -245,10 +250,12 @@ attr_index_config(
 						   preamble, officialOID, index_rules[j] );
 					slapi_ch_free((void**)&preamble);
 				    }
-				} else {
+				} else if (!slapi_matchingrule_is_ordering(index_rules[j], attrsyntax_oid)) {
 				    LDAPDebug (LDAP_DEBUG_ANY, "%s: line %d: "
-					       "unknown index rule \"%s\" (ignored)\n",
+					       "unknown or invalid matching rule \"%s\" in index configuration (ignored)\n",
 					       fname, lineno, index_rules[j] );
+				} else { /* assume builtin and use compare fn provided by syntax plugin */
+				    need_compare_fn = 1;
 				}
 				{/* It would improve speed to save the indexer, for future use.
 				    But, for simplicity, we destroy it now: */
@@ -269,13 +276,8 @@ attr_index_config(
 			    }
 			}
 		}
-#if 0    /* seems to not matter -- INDEX_FROMINIT is checked nowhere else */
-		if ( init ) {
-			a->ai_indexmask |= INDEX_FROMINIT;
-                        a->ai_indexmask &= ~INDEX_OFFLINE;
-		}
-#endif
 
+		slapi_ch_free_string(&attrsyntax_oid);
 		/* initialize the IDL code's private data */
 		return_value = idl_init_private(be, a);
 		if (0 != return_value) {
@@ -285,6 +287,26 @@ attr_index_config(
 			exit( 1 );
 		}
 
+		/* if user didn't specify an ordering rule in the index config,
+		   see if the schema def for the attr defines one */
+		if (!need_compare_fn) {
+			asyntaxinfo *asi = attr_syntax_get_by_name( a->ai_type );
+			if (asi && asi->asi_mr_ordering) {
+			 	need_compare_fn = 1;
+			}
+			attr_syntax_return( asi );
+		}
+
+		if (need_compare_fn) {
+			int rc = plugin_call_syntax_get_compare_fn( a->ai_plugin, &a->ai_key_cmp_fn );
+			if (rc != LDAP_SUCCESS) {
+			    LDAPDebug(LDAP_DEBUG_ANY,
+				      "The attribute [%s] does not have a valid ORDERING matching rule\n",
+				      a->ai_type, 0, 0);
+				a->ai_key_cmp_fn = NULL;
+			}
+		}
+
 		if ( avl_insert( &inst->inst_attrs, a, ainfo_cmp, ainfo_dup ) != 0 ) {
 			/* duplicate - existing version updated */
             attrinfo_delete(&a);

ldap/servers/slapd/back-ldbm/proto-back-ldbm.h

@@ -141,7 +141,7 @@ int dblayer_delete_database(struct ldbminfo *li);
 int dblayer_database_size(struct ldbminfo *li, unsigned int *size);
 int dblayer_terminate(struct ldbminfo *li);
 int dblayer_close_indexes(backend *be);
-int dblayer_open_file(backend *be, char* indexname, int create, int index_flags, DB **ppDB);
+int dblayer_open_file(backend *be, char* indexname, int create, struct attrinfo *ai, DB **ppDB);
 int dblayer_close_file(DB *db);
 void dblayer_sys_pages(size_t *pagesize, size_t *pages, size_t *procpages, size_t *availpages);
 int dblayer_is_cachesize_sane(size_t *cachesize);

ldap/servers/slapd/match.c

@@ -270,5 +270,26 @@ int slapi_matchingrule_unregister(char *oid)
     return(0);
 }
 
+/*
+  See if a matching rule for this name or OID
+  is registered and is an ORDERING matching rule that applies
+  to the given syntax.
+*/
+int slapi_matchingrule_is_ordering(const char *oid_or_name, const char *syntax_oid)
+{
+    struct matchingRuleList *mrl=NULL;
+    for (mrl = g_get_global_mrl(); mrl != NULL; mrl = mrl->mrl_next) {
+        if (mrl->mr_entry->mr_name && !strcasecmp(oid_or_name, mrl->mr_entry->mr_name)) {
+            return (mrl->mr_entry->mr_name &&
+                    PL_strcasestr(mrl->mr_entry->mr_name, "ordering") &&
+                    !strcmp(mrl->mr_entry->mr_syntax, syntax_oid));
+        }
+        if (mrl->mr_entry->mr_oid && !strcmp(oid_or_name, mrl->mr_entry->mr_oid)) {
+            return (mrl->mr_entry->mr_name &&
+                    PL_strcasestr(mrl->mr_entry->mr_name, "ordering") &&
+                    !strcmp(mrl->mr_entry->mr_syntax, syntax_oid));
+        }
+    }
 
-
+    return 0;
+}

ldap/servers/slapd/slap.h

@@ -616,15 +616,6 @@ struct objclass {
 	struct objclass		*oc_next;
 };
 
-typedef struct slapi_matchingRuleEntry {
-    char *mr_oid;
-    char *mr_oidalias;
-    char *mr_name;
-    char *mr_desc;
-    char *mr_syntax;
-    int mr_obsolete;
-} slapi_matchingRuleEntry;
- 
 struct matchingRuleList {
     Slapi_MatchingRuleEntry *mr_entry;
     struct matchingRuleList *mrl_next;

ldap/servers/slapd/slapi-plugin.h

@@ -140,7 +140,6 @@ typedef struct slapi_attr		Slapi_Attr;
 typedef struct slapi_value  		Slapi_Value;
 typedef struct slapi_value_set  	Slapi_ValueSet;
 typedef struct slapi_filter		Slapi_Filter;
-typedef struct slapi_matchingRuleEntry	Slapi_MatchingRuleEntry;
 typedef struct backend			Slapi_Backend;
 typedef struct _guid_t			Slapi_UniqueID;
 typedef struct op			Slapi_Operation;
@@ -645,6 +644,16 @@ int slapi_berval_cmp(const struct berval* L, const struct berval* R);
 #define SLAPI_BERVAL_EQ(L,R) ((L)->bv_len == (R)->bv_len && \
         ! memcmp ((L)->bv_val, (R)->bv_val, (L)->bv_len))
 
+typedef struct slapi_matchingRuleEntry {
+    char *mr_oid;
+    char *mr_oidalias;
+    char *mr_name;
+    char *mr_desc;
+    char *mr_syntax;
+    int mr_obsolete;
+} slapi_matchingRuleEntry;
+typedef struct slapi_matchingRuleEntry	Slapi_MatchingRuleEntry;
+
 Slapi_MatchingRuleEntry *slapi_matchingrule_new(void);
 void slapi_matchingrule_free(Slapi_MatchingRuleEntry **mrEntry,
                              int freeMembers);
@@ -652,6 +661,7 @@ int slapi_matchingrule_get(Slapi_MatchingRuleEntry *mr, int arg, void *value);
 int slapi_matchingrule_set(Slapi_MatchingRuleEntry *mr, int arg, void *value);
 int slapi_matchingrule_register(Slapi_MatchingRuleEntry *mrEntry);
 int slapi_matchingrule_unregister(char *oid);
+int slapi_matchingrule_is_ordering(const char *oid_or_name, const char *syntax_oid);
 
 /*
  * access control