linnaea/README.md

## README.md

      
    Raw
  

              README.md
            
          
    Cassandra backend for FreeRADIUS.
Please make sure that in accounting section of the site config, cassacctsupplement appears before cassacct, cassauth should also be added to the accounting section to enforce Simultaneous-Use checks.

  
## all.mk
TARGET		= rlm_sql_cassandra.a
SOURCES		= rlm_sql_cassandra.c

SRC_CFLAGS	= -I${top_srcdir}/src/modules/rlm_sql
TGT_LDLIBS      = -lcassandra

## rlm_sql_cassandra.c
/*
This is free and unencumbered software released into the public domain.

Anyone is free to copy, modify, publish, use, compile, sell, or
distribute this software, either in source code form or as a compiled
binary, for any purpose, commercial or non-commercial, and by any
means.

In jurisdictions that recognize copyright laws, the author or authors
of this software dedicate any and all copyright interest in the
software to the public domain. We make this dedication for the benefit
of the public at large and to the detriment of our heirs and
successors. We intend this dedication to be an overt act of
relinquishment in perpetuity of all present and future rights to this
software under copyright law.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
IN NO EVENT SHALL THE AUTHORS BE LIABLE FOR ANY CLAIM, DAMAGES OR
OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE,
ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
OTHER DEALINGS IN THE SOFTWARE.

For more information, please refer to <http://unlicense.org>
*/

#include <freeradius-devel/radiusd.h>
#include <freeradius-devel/rad_assert.h>

#include <sys/stat.h>

#include <cassandra.h>

#include "rlm_sql.h"

#define RLM_SQL_CASS_LOG_BUFFER 128

typedef struct rlm_sql_cassandra_conn {
	      CassSession  *sess;
	      CassCluster  *cluster;
	const CassResult   *res;
	      CassIterator *resIter;
	      char        **resBuffer;
} rlm_sql_cassandra_conn_t;

typedef struct rlm_sql_cassandra_config {
	char const		*consistency_str;
	CassConsistency	consistency;
	sql_log_entry_t logBuffer[RLM_SQL_CASS_LOG_BUFFER];
	unsigned int	logRingPosition;
} rlm_sql_cassandra_config_t;

static const CONF_PARSER driver_config[] = {
	{ "consistency", FR_CONF_OFFSET(PW_TYPE_STRING, rlm_sql_cassandra_config_t, consistency_str), "quorum" },
	{NULL, -1, 0, NULL, NULL}
};

static const FR_NAME_NUMBER consistency_levels[] = {
	{ "any",	CASS_CONSISTENCY_ANY	},
	{ "one",	CASS_CONSISTENCY_ONE	},
	{ "two",	CASS_CONSISTENCY_TWO	},
	{ "three",	CASS_CONSISTENCY_THREE	},
	{ "quorum",	CASS_CONSISTENCY_QUORUM	},
	{ "all",	CASS_CONSISTENCY_ALL	},
	{ "each_quorum",	CASS_CONSISTENCY_EACH_QUORUM	},
	{ "local_quorum",	CASS_CONSISTENCY_LOCAL_QUORUM	},
	{ "local_one",		CASS_CONSISTENCY_LOCAL_ONE		},
	{ NULL, 0 }
};

void casslog(const CassLogMessage* message, void* data) {
	rlm_sql_cassandra_config_t *conf = data;
	int logPosition = __sync_fetch_and_add(&conf->logRingPosition, 1) % RLM_SQL_CASS_LOG_BUFFER;
	switch(message->severity) {
		case CASS_LOG_CRITICAL:
		case CASS_LOG_ERROR:
			conf->logBuffer[logPosition].type = L_ERR;
			break;
		case CASS_LOG_WARN:
			conf->logBuffer[logPosition].type = L_WARN;
			break;
		case CASS_LOG_INFO:
		case CASS_LOG_DISABLED:
		case CASS_LOG_LAST_ENTRY:
			conf->logBuffer[logPosition].type = L_INFO;
			break;
		case CASS_LOG_DEBUG:
		case CASS_LOG_TRACE:
			conf->logBuffer[logPosition].type = L_DBG;
			break;
	}
	if(conf->logBuffer[logPosition].msg != NULL) talloc_free(conf->logBuffer[logPosition].msg);
	MEM(conf->logBuffer[logPosition].msg = talloc_asprintf(conf, "rlm_sql_cassandra: [drv][%" PRId64 "][%d] In file %s, function %s, on line %d: %s", (int64_t)message->time_ms, (int)message->severity, message->file, message->function, message->line, message->message));
}

static int _mod_destructor(rlm_sql_cassandra_config_t *conf)
{
	DEBUG2("rlm_sql_cassandra: Module shut down");
	cass_log_cleanup();
	return 0;
}

static int mod_instantiate(CONF_SECTION *conf, rlm_sql_config_t *config)
{
	static bool version_done = false;

	rlm_sql_cassandra_config_t *driver;
	int consistency;

	if (!version_done) {
		version_done = true;

		INFO("rlm_sql_cassandra: compiled against Cassandra cpp-driver %d.%d.%d%s", CASS_VERSION_MAJOR, CASS_VERSION_MINOR, CASS_VERSION_PATCH, CASS_VERSION_SUFFIX);
	}

	MEM(driver = config->driver = talloc_zero(config, rlm_sql_cassandra_config_t));
	talloc_set_destructor(driver, _mod_destructor);

	if (cf_section_parse(conf, driver, driver_config) < 0) {
		return -1;
	}

	consistency = fr_str2int(consistency_levels, driver->consistency_str, -1);
	if (consistency < 0) {
		ERROR("rlm_sql_cassandra: Invalid consistency level %s", driver->consistency_str);
		return -1;
	}
	driver->consistency = (CassConsistency)consistency;
	driver->logRingPosition = 0;
	for(int i = 0; i < RLM_SQL_CASS_LOG_BUFFER; i++) {
		driver->logBuffer[i].msg = NULL;
	}
	cass_log_set_level(CASS_LOG_INFO);
	cass_log_set_callback(casslog, driver);

	return 0;
}

static int _sql_socket_destructor(rlm_sql_cassandra_conn_t *conn)
{
	DEBUG2("rlm_sql_cassandra: Socket destructor called, closing socket");

	if (conn->resIter) {
		cass_iterator_free(conn->resIter);
	}
	if (conn->res) {
		cass_result_free(conn->res);
	}
	if (conn->sess) {
		cass_session_free(conn->sess);
	}
	if(conn->cluster) {
		cass_cluster_free(conn->cluster);
	}

	return 0;
}

static sql_rcode_t sql_socket_init(rlm_sql_handle_t *handle, rlm_sql_config_t *config)
{
	rlm_sql_cassandra_conn_t *conn;
	CassCluster *cluster;
	CassSession *sess;
	CassFuture *future;
	CassError error;

	MEM(conn = handle->conn = talloc_zero(handle, rlm_sql_cassandra_conn_t));
	talloc_set_destructor(conn, _sql_socket_destructor);

	DEBUG4("rlm_sql_cassandra: Configuring CassCluster");
	cluster = conn->cluster = cass_cluster_new();
	cass_cluster_set_contact_points(cluster, config->sql_server);
	cass_cluster_set_port(cluster, atoi(config->sql_port));
	cass_cluster_set_connect_timeout(cluster, config->connect_timeout_ms);
	cass_cluster_set_request_timeout(cluster, config->query_timeout);
	cass_cluster_set_credentials(cluster, config->sql_login, config->sql_password);

	DEBUG4("rlm_sql_cassandra: Connecting to Cassandra Cluster");
	sess = conn->sess = cass_session_new();
	future = cass_session_connect_keyspace(sess, cluster, config->sql_db);
	error = cass_future_error_code(future);
	if(error != CASS_OK) {
		const char *msg; size_t msgLen;
		cass_future_error_message(future, &msg, &msgLen);
		ERROR("rlm_sql_cassandra: Unable to connect: [%x] %s", (int)error, msg);
		cass_future_free(future);
		return RLM_SQL_ERROR;
	}
	cass_future_free(future);

	DEBUG4("rlm_sql_cassandra: Connected to Cassandra Cluster");

	return RLM_SQL_OK;
}

static sql_rcode_t sql_query(rlm_sql_handle_t *handle, rlm_sql_config_t *config, char const *query)
{
	rlm_sql_cassandra_conn_t *conn = handle->conn;
	rlm_sql_cassandra_config_t *conf = config->driver;
	CassStatement *stmt;
	CassFuture *future;
	CassError error;

	if(query[0] == ';') return RLM_SQL_OK;

	stmt = cass_statement_new(query, 0);
	cass_statement_set_consistency(stmt, conf->consistency);

	future = cass_session_execute(conn->sess, stmt);
	cass_statement_free(stmt);
	error = cass_future_error_code(future);
	if(error != CASS_OK) {
		const char *msg; size_t msgLen;
		cass_future_error_message(future, &msg, &msgLen);
		WARN("rlm_sql_cassandra: query failed: [%x] %s", (int)error, msg);
		DEBUG("rlm_sql_cassandra: the query was: %s", query);
		cass_future_free(future);
		switch(error) {
			case CASS_ERROR_SERVER_SYNTAX_ERROR:
			case CASS_ERROR_SERVER_INVALID_QUERY:
				return RLM_SQL_QUERY_INVALID;
			default:
				return RLM_SQL_ERROR;
		}
	}
	conn->res = cass_future_get_result(future);
	cass_future_free(future);
	if(!conn->res) DEBUG4("rlm_sql_cassandra: query doesn't return a result.");

	return RLM_SQL_OK;
}

static sql_rcode_t sql_store_result(rlm_sql_handle_t * handle, UNUSED rlm_sql_config_t *config)
{
	return RLM_SQL_OK;
}

static int sql_num_fields(rlm_sql_handle_t *handle, UNUSED rlm_sql_config_t *config)
{
	rlm_sql_cassandra_conn_t *conn = handle->conn;
	if(conn->res) {
		return cass_result_column_count(conn->res);
	}
	WARN("rlm_sql_cassandra: sql_num_fields called without an outstanding query result.");
	return 0;
}

static int sql_num_rows(rlm_sql_handle_t *handle, UNUSED rlm_sql_config_t *config)
{
	rlm_sql_cassandra_conn_t *conn = handle->conn;
	if(conn->res) {
		return cass_result_row_count(conn->res);
	}
	WARN("rlm_sql_cassandra: sql_num_rows called without an outstanding query result.");
	return 0;
}

#define RLM_CASS_ERR_DATA_RETRIVE(type) {\
				const char *colName; size_t colNameLen;\
				if(cass_result_column_name(conn->res, i, &colName, &colNameLen) != CASS_OK) colName = "error";\
				WARN("rlm_sql_cassandra: failed to retrive " type " data at column (%d)%s", i, colName);\
				conn->resBuffer[i] = "";\
				break;}
static sql_rcode_t sql_fetch_row(rlm_sql_handle_t *handle, rlm_sql_config_t *config)
{
	rlm_sql_cassandra_conn_t *conn = handle->conn;
	handle->row = NULL;
	if(conn->res && !conn->resIter) {
		conn->resIter = cass_iterator_from_result(conn->res);
	}
	if(conn->res && conn->resIter) {
		if(cass_iterator_next(conn->resIter)) {
			if(conn->resBuffer != NULL) talloc_free(conn->resBuffer);
			const CassRow* row = cass_iterator_get_row(conn->resIter);
			int fields = sql_num_fields(handle, config);
			MEM(handle->row = conn->resBuffer = talloc_array(handle, char *, fields));
			for(int i = 0; i < fields; i++) {
				const CassValue *value = cass_row_get_column(row, i);
				conn->resBuffer[i] = NULL;
				if(cass_value_is_null(value) == cass_true) {
					continue;
				}
				CassValueType type = cass_value_type(value);
				switch(type) {
					case CASS_VALUE_TYPE_ASCII:
					case CASS_VALUE_TYPE_TEXT:
					case CASS_VALUE_TYPE_VARCHAR: {
						const char *str; size_t strLen;
						if(cass_value_get_string(value, &str, &strLen) != CASS_OK) {
							RLM_CASS_ERR_DATA_RETRIVE("string");
						}
						MEM(conn->resBuffer[i] = talloc_array(conn->resBuffer, char, strLen+1));
						memcpy(conn->resBuffer[i], str, strLen);
						conn->resBuffer[i][strLen] = '\0';
						break;
					}
					case CASS_VALUE_TYPE_BOOLEAN: {
						cass_bool_t bv;
						if(cass_value_get_bool(value, &bv) != CASS_OK) {
							RLM_CASS_ERR_DATA_RETRIVE("bool");
						}
						MEM(conn->resBuffer[i] = talloc_zero_array(conn->resBuffer, char, 2));
						if(bv == cass_false) {
							conn->resBuffer[i][0] = '0';
						} else {
							conn->resBuffer[i][0] = '1';
						}
						break;
					}
					case CASS_VALUE_TYPE_INT: {
						cass_int32_t i32v;
						if(cass_value_get_int32(value, &i32v) != CASS_OK) {
							RLM_CASS_ERR_DATA_RETRIVE("int32");
						}
						MEM(conn->resBuffer[i] = talloc_asprintf(conn->resBuffer, "%"PRId32, (int32_t)i32v));
						break;
					}
					case CASS_VALUE_TYPE_TIMESTAMP:
					case CASS_VALUE_TYPE_BIGINT: {
						cass_int64_t i64v;
						if(cass_value_get_int64(value, &i64v) != CASS_OK) {
							RLM_CASS_ERR_DATA_RETRIVE("int64");
						}
						MEM(conn->resBuffer[i] = talloc_asprintf(conn->resBuffer, "%"PRId64, (int64_t)i64v));
						break;
					}
					case CASS_VALUE_TYPE_UUID:
					case CASS_VALUE_TYPE_TIMEUUID: {
						CassUuid uuid;
						if(cass_value_get_uuid(value, &uuid) != CASS_OK) {
							RLM_CASS_ERR_DATA_RETRIVE("UUID");
						}
						MEM(conn->resBuffer[i] = talloc_array(conn->resBuffer, char, CASS_UUID_STRING_LENGTH));
						cass_uuid_string(uuid, conn->resBuffer[i]);
						break;
					}
					default: {
						const char *colName; size_t colNameLen;
						if(cass_result_column_name(conn->res, i, &colName, &colNameLen) != CASS_OK) colName = "error";
						WARN("rlm_sql_cassandra: column (%d)%s: Unsupported type %d", i, colName, (int)type);
						break;
					}
				}
				DEBUG4("rlm_sql_cassandra: row[%d] => %s", i, conn->resBuffer[i] ? conn->resBuffer[i] : "(null)");
			}
		} else DEBUG4("rlm_sql_cassandra: no more results.");
	} else WARN("rlm_sql_cassandra: sql_fetch_row called without an outstanding query result.");
	return RLM_SQL_OK;
}

static sql_rcode_t sql_free_result(rlm_sql_handle_t * handle, UNUSED rlm_sql_config_t *config)
{
	rlm_sql_cassandra_conn_t *conn = handle->conn;

	if(conn->resBuffer != NULL) talloc_free(conn->resBuffer);
	if(conn->resIter != NULL) cass_iterator_free(conn->resIter);
	if(conn->res != NULL) cass_result_free(conn->res);

	conn->resBuffer = NULL;
	conn->resIter = NULL;
	conn->res = NULL;

	return RLM_SQL_OK;
}

static size_t sql_error(TALLOC_CTX *ctx, sql_log_entry_t out[], size_t outlen, UNUSED rlm_sql_handle_t *handle, rlm_sql_config_t *config)
{
	rlm_sql_cassandra_config_t	*driver = config->driver;
	int fillSize = outlen > RLM_SQL_CASS_LOG_BUFFER ? RLM_SQL_CASS_LOG_BUFFER : outlen;
	int bufferFilled = driver->logRingPosition;
	fillSize = fillSize > bufferFilled ? bufferFilled : fillSize;

	for(int i = 0; i < fillSize; i++) {
		sql_log_entry_t entry = driver->logBuffer[(bufferFilled - fillSize + i) % RLM_SQL_CASS_LOG_BUFFER];
		size_t msgLen = strlen(entry.msg)+1;
		out[i].type = entry.type;
		out[i].msg = talloc_array(ctx, char, msgLen);
		memcpy(out[i].msg, entry.msg, msgLen);
	}

	return fillSize;
}

static sql_rcode_t sql_finish_query(rlm_sql_handle_t *handle, rlm_sql_config_t *config)
{
	return sql_free_result(handle, config);
}

//Unsupported operation on Cassandra Cluster, fake one.
static int sql_affected_rows(UNUSED rlm_sql_handle_t * handle, UNUSED rlm_sql_config_t *config) { return 1; }

/* Exported to rlm_sql */
extern rlm_sql_module_t rlm_sql_cassandra;
rlm_sql_module_t rlm_sql_cassandra = {
	.name				= "rlm_sql_cassandra",
	.mod_instantiate		= mod_instantiate,
	.sql_socket_init		= sql_socket_init,
	.sql_query				= sql_query,
	.sql_select_query		= sql_query,
	.sql_store_result		= sql_store_result,
	.sql_num_fields			= sql_num_fields,
	.sql_num_rows			= sql_num_rows,
	.sql_fetch_row			= sql_fetch_row,
	.sql_free_result		= sql_free_result,
	.sql_error				= sql_error,
	.sql_finish_query		= sql_finish_query,
	.sql_finish_select_query	= sql_finish_query,
	.sql_affected_rows		= sql_affected_rows
};

## schema.cql
CREATE KEYSPACE radius WITH REPLICATION = { 'class' : 'SimpleStrategy', 'replication_factor' : 1 };
USE radius;

CREATE TABLE nas (
  id uuid PRIMARY KEY,
  nasname text,
  shortname text,
  type text,
  secret text,
  server text
);

CREATE TABLE radpostauth (
  username text,
  authdate bigint,
  reply text,
  PRIMARY KEY (username, authdate)
) WITH CLUSTERING ORDER BY (authdate ASC);

CREATE TABLE radusergroup (
  username text,
  priority int,
  groupname text,
  PRIMARY KEY (username, priority)
) WITH CLUSTERING ORDER BY (priority ASC);

CREATE TABLE radreply (
  id uuid,
  username text,
  attribute text,
  op text,
  value text,
  PRIMARY KEY (username, attribute)
);

CREATE TABLE radgroupreply (
  id uuid,
  groupname text,
  attribute text,
  op text,
  value text,
  PRIMARY KEY (groupname, attribute)
);

CREATE TABLE radgroupcheck (
  id uuid,
  groupname text,
  attribute text,
  op text,
  value text,
  PRIMARY KEY (groupname, attribute)
);

CREATE TABLE radcheck (
  id uuid,
  username text,
  attribute text,
  op text,
  value text,
  PRIMARY KEY (username, attribute)
);

CREATE TABLE radacct (
  id uuid,
  sessionid text,
  username text,
  realm text,
  nasipaddress text,
  nasportid text,
  nasporttype text,
  starttime bigint,
  updatetime bigint,
  stoptime bigint,
  sessiontime int,
  authentic text,
  connectinfo_start text,
  connectinfo_stop text,
  inputoctetslo bigint,
  inputoctetshi bigint,
  outputoctetslo bigint,
  outputoctetshi bigint,
  calledstationid text,
  callingstationid text,
  servicetype text,
  terminatecause text,
  framedprotocol text,
  framedipaddress text,
  PRIMARY KEY (username, nasipaddress, sessionid)
);

CREATE TABLE radnasflip (
  nasipaddress text,
  timestamp bigint,
  PRIMARY KEY (timestamp, nasipaddress)
);

CREATE TABLE radsess (
  id uuid,
  sessionid text,
  username text,
  nasipaddress text,
  nasportid text,
  callingstationid text,
  framedprotocol text,
  framedipaddress text,
  PRIMARY KEY (nasipaddress, username, sessionid)
);
CREATE INDEX ON radsess(username);

## sql.conf
sql cassauth {
        driver = "rlm_sql_cassandra"
        server = "localhost"
        port = 9042
        login = ""
        password = ""
        radius_db = "radius"
        query_timeout = 15000
        safe_characters = "@abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789.-_: /[]"
        sql_user_name = "%{%{Stripped-User-Name}:-%{%{User-Name}:-Guest}}"
        default_user_profile = "DEFAULT"

        cassandra {
                consistency = "local_quorum"
        }

        pool {
                start = 1
                min = 1
                max = ${thread[pool].max_servers}
                spare = 2
                uses = 0
                lifetime = 0
                idle_timeout = 60
        }

        read_clients = no
        postauth_table = "radpostauth"
        authcheck_table = "radcheck"
        groupcheck_table = "radgroupcheck"
        authreply_table = "radreply"
        groupreply_table = "radgroupreply"
        usergroup_table = "radusergroup"
        client_table = "nas"
        session_table = "radsess"
        session_ttl = "USING TTL 3600"
        postauth_ttl = "USING TTL 6307200"

        simul_count_query = "\
                SELECT COUNT(*) \
                FROM ${session_table} \
                WHERE username = '%{SQL-User-Name}'"

        simul_verify_query = "\
                SELECT  id, sessionid, username, nasipaddress, nasportid, framedipaddress, \
                        callingstationid, framedprotocol \
                FROM ${session_table} \
                WHERE username = '%{SQL-User-Name}'"

        client_query = "\
                SELECT id, nasname, shortname, type, secret, server \
                FROM ${client_table}"

        authorize_check_query = "\
                SELECT id, username, attribute, value, op \
                FROM ${authcheck_table} \
                WHERE username = '%{SQL-User-Name}'"

        authorize_reply_query = "\
                SELECT id, username, attribute, value, op \
                FROM ${authreply_table} \
                WHERE username = '%{SQL-User-Name}'"

        group_membership_query = "\
                SELECT groupname \
                FROM ${usergroup_table} \
                WHERE username = '%{SQL-User-Name}' \
                ORDER BY priority"

        authorize_group_check_query = "\
                SELECT id, groupname, attribute, value, op \
                FROM ${groupcheck_table} \
                WHERE groupname = '%{Sql-Group}'"

        authorize_group_reply_query = "\
                SELECT id, groupname, attribute, value, op \
                FROM ${groupreply_table} \
                WHERE groupname = '%{Sql-Group}'"

        accounting {
                reference = "%{tolower:type.%{Acct-Status-Type}.query}"

                type {
                        accounting-on {
                                query = "DELETE FROM ${....session_table} WHERE nasipaddress = '%{NAS-IP-Address}'"
                        }

                        accounting-off {
                                query = "${..accounting-on.query}"
                        }

                        start {
                                query = "\
                                        INSERT INTO ${....session_table} ( \
                                                id, sessionid, username, \
                                                nasipaddress, nasportid, \
                                                callingstationid, \
                                                framedprotocol, framedipaddress \
                                        ) VALUES ( \
                                                uuid(), '%{Acct-Session-Id}', '%{SQL-User-Name}', \
                                                '%{NAS-IP-Address}', '%{NAS-Port}', \
                                                '%{Calling-Station-Id}', \
                                                '%{Framed-Protocol}', '%{Framed-IP-Address}' \
                                        ) ${....session_ttl};"
                        }

                        interim-update {
                                query = "${..start.query}"
                        }

                        stop {
                                query = "DELETE FROM ${....session_table} \
                                        WHERE nasipaddress = '%{NAS-IP-Address}' \
                                        AND   username     = '%{SQL-User-Name}' \
                                        AND   sessionid    = '%{Acct-Session-Id}'"
                        }
                }
        }

        post-auth {
                query = "\
                        INSERT INTO ${..postauth_table} \
                                (username, reply, authdate) \
                        VALUES ( \
                                '%{SQL-User-Name}', \
                                '%{reply:Packet-Type}', \
                                unixTimestampOf(now()) \
                        ) ${..postauth_ttl};"
        }
}

sql cassacct {
        driver = "rlm_sql_cassandra"
        server = "localhost"
        port = 9042
        login = ""
        password = ""
        radius_db = "radius"
        query_timeout = 15000
        safe_characters = "@abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789.-_: /[]"
        sql_user_name = "%{%{Stripped-User-Name}:-%{%{User-Name}:-Guest}}"
        default_user_profile = "DEFAULT"

        cassandra {
                consistency = "local_quorum"
        }

        pool {
                start = 1
                min = 1
                max = ${thread[pool].max_servers}
                spare = 2
                uses = 0
                lifetime = 0
                idle_timeout = 60
        }

        acct_ttl = "USING TTL 6307200"
        acct_table = "radacct"
        flip_table = "radnasflip"

        accounting {
                reference = "%{tolower:type.%{Acct-Status-Type}.query}"

                type {
                        accounting-on {
                                query = "\
                                        INSERT INTO ${....flip_table} (nasipaddress, timestamp) \
                                        VALUES ('%{NAS-IP-Address}', %{integer:Event-Timestamp}) \
                                        ${....acct_ttl};"
                        }

                        accounting-off {
                                query = "${..accounting-on.query}"
                        }

                        start {
                                query = "\
                                        INSERT INTO ${....acct_table} ( \
                                                id, sessionid, username, realm, \
                                                nasipaddress, nasportid, nasporttype, \
                                                starttime, updatetime, stoptime, sessiontime, \
                                                authentic, connectinfo_start, connectinfo_stop, \
                                                inputoctetslo, inputoctetshi, outputoctetslo, outputoctetshi, \
                                                calledstationid, callingstationid, servicetype, \
                                                terminatecause, framedprotocol, framedipaddress \
                                        ) VALUES ( \
                                                uuid(), '%{Acct-Session-Id}', '%{SQL-User-Name}', '%{Realm}', \
                                                '%{NAS-IP-Address}', '%{NAS-Port}', '%{NAS-Port-Type}', \
                                                %{integer:Event-Timestamp}, %{integer:Event-Timestamp}, \
                                                0, 0, \
                                                '%{Acct-Authentic}', '%{Connect-Info}', '', \
                                                0, 0, 0, 0, \
                                                '%{Called-Station-Id}', '%{Calling-Station-Id}', '%{Service-Type}', \
                                                '', '%{Framed-Protocol}', '%{Framed-IP-Address}' \
                                        ) ${....acct_ttl};"
                        }

                        interim-update {
                                query = "\
                                        INSERT INTO ${....acct_table} ( \
                                                sessionid, username, nasipaddress, \
                                                updatetime, sessiontime, \
                                                framedipaddress, \
                                                inputoctetshi, inputoctetslo, \
                                                outputoctetshi, outputoctetslo \
                                        ) VALUES ( \
                                                '%{Acct-Session-Id}', '%{SQL-User-Name}', '%{NAS-IP-Address}', \
                                                %{integer:Event-Timestamp}, %{Acct-Session-Time}, \
                                                '%{Framed-IP-Address}', \
                                                %{%{Acct-Input-Gigawords}:-0}, %{%{Acct-Input-Octets}:-0}, \
                                                %{%{Acct-Output-Gigawords}:-0}, %{%{Acct-Output-Octets}:-0} \
                                        ) ${....acct_ttl};"
                        }

                        stop {
                                query = "\
                                        INSERT INTO ${....acct_table} ( \
                                                sessionid, username, nasipaddress, \
                                                updatetime, sessiontime, stoptime, \
                                                inputoctetshi, inputoctetslo, \
                                                outputoctetshi, outputoctetslo, \
                                                connectinfo_stop, terminatecause \
                                        ) VALUES ( \
                                                '%{Acct-Session-Id}', '%{SQL-User-Name}', '%{NAS-IP-Address}', \
                                                %{integer:Event-Timestamp}, %{Acct-Session-Time}, %{integer:Event-Timestamp}, \
                                                %{%{Acct-Input-Gigawords}:-0}, %{%{Acct-Input-Octets}:-0}, \
                                                %{%{Acct-Output-Gigawords}:-0}, %{%{Acct-Output-Octets}:-0}, \
                                                '%{Connect-Info}', '%{Acct-Terminate-Cause}' \
                                        ) ${....acct_ttl};"
                        }
                }
        }
}

sql cassacctsupplement {
        driver = "rlm_sql_cassandra"
        server = "localhost"
        port = 9042
        login = ""
        password = ""
        radius_db = "radius"
        query_timeout = 15000
        safe_characters = "@abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789.-_: /[]"
        sql_user_name = "%{%{Stripped-User-Name}:-%{%{User-Name}:-Guest}}"
        default_user_profile = "DEFAULT"

        cassandra {
                consistency = "local_quorum"
        }

        pool {
                start = 1
                min = 1
                max = ${thread[pool].max_servers}
                spare = 2
                uses = 0
                lifetime = 0
                idle_timeout = 60
        }

        acct_ttl = "USING TTL 6307200"
        acct_table = "radacct"

        accounting {
                reference = "%{tolower:type.%{Acct-Status-Type}.query}"

                type {
                        interim-update {
                                query = "\
                                        INSERT INTO radacct ( \
                                                id, sessionid, username, realm, \
                                                nasipaddress, nasportid, nasporttype, \
                                                starttime, stoptime, \
                                                authentic, connectinfo_start, connectinfo_stop, \
                                                calledstationid, callingstationid, servicetype, \
                                                terminatecause, framedprotocol \
                                        ) VALUES ( \
                                                uuid(), '%{Acct-Session-Id}', '%{SQL-User-Name}', '%{Realm}', \
                                                '%{NAS-IP-Address}', '%{NAS-Port}', '%{NAS-Port-Type}', \
                                                0, 0, \
                                                '%{Acct-Authentic}', '%{Connect-Info}', '', \
                                                '%{Called-Station-Id}', '%{Calling-Station-Id}', '%{Service-Type}', \
                                                '', '%{Framed-Protocol}' \
                                        ) IF NOT EXISTS ${....acct_ttl};"
                        }

                        stop {
                                query = "\
                                        INSERT INTO radacct ( \
                                                id, sessionid, username, realm, \
                                                nasipaddress, nasportid, nasporttype, \
                                                starttime, \
                                                authentic, connectinfo_start, \
                                                calledstationid, callingstationid, servicetype, \
                                                framedprotocol, framedipaddress \
                                        ) VALUES ( \
                                                uuid(), '%{Acct-Session-Id}', '%{SQL-User-Name}', '%{Realm}', \
                                                '%{NAS-IP-Address}', '%{NAS-Port}', '%{NAS-Port-Type}', \
                                                0, \
                                                '%{Acct-Authentic}', '', \
                                                '%{Called-Station-Id}', '%{Calling-Station-Id}', '%{Service-Type}', \
                                                '%{Framed-Protocol}', '%{Framed-IP-Address}' \
                                        ) IF NOT EXISTS ${....acct_ttl};"
                        }

                        start {
                                query = ";"
                        }
                        accounting-on {
                                query = ";"
                        }
                        accounting-off {
                                query = ";"
                        }
                }
        }
}
	TARGET = rlm_sql_cassandra.a
	SOURCES = rlm_sql_cassandra.c

	SRC_CFLAGS = -I${top_srcdir}/src/modules/rlm_sql
	TGT_LDLIBS = -lcassandra
	/*
	This is free and unencumbered software released into the public domain.

	Anyone is free to copy, modify, publish, use, compile, sell, or
	distribute this software, either in source code form or as a compiled
	binary, for any purpose, commercial or non-commercial, and by any
	means.

	In jurisdictions that recognize copyright laws, the author or authors
	of this software dedicate any and all copyright interest in the
	software to the public domain. We make this dedication for the benefit
	of the public at large and to the detriment of our heirs and
	successors. We intend this dedication to be an overt act of
	relinquishment in perpetuity of all present and future rights to this
	software under copyright law.

	THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
	EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
	MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
	IN NO EVENT SHALL THE AUTHORS BE LIABLE FOR ANY CLAIM, DAMAGES OR
	OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE,
	ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
	OTHER DEALINGS IN THE SOFTWARE.

	For more information, please refer to <http://unlicense.org>
	*/

	#include <freeradius-devel/radiusd.h>
	#include <freeradius-devel/rad_assert.h>

	#include <sys/stat.h>

	#include <cassandra.h>

	#include "rlm_sql.h"

	#define RLM_SQL_CASS_LOG_BUFFER 128

	typedef struct rlm_sql_cassandra_conn {
	CassSession *sess;
	CassCluster *cluster;
	const CassResult *res;
	CassIterator *resIter;
	char **resBuffer;
	} rlm_sql_cassandra_conn_t;

	typedef struct rlm_sql_cassandra_config {
	char const *consistency_str;
	CassConsistency consistency;
	sql_log_entry_t logBuffer[RLM_SQL_CASS_LOG_BUFFER];
	unsigned int logRingPosition;
	} rlm_sql_cassandra_config_t;

	static const CONF_PARSER driver_config[] = {
	{ "consistency", FR_CONF_OFFSET(PW_TYPE_STRING, rlm_sql_cassandra_config_t, consistency_str), "quorum" },
	{NULL, -1, 0, NULL, NULL}
	};

	static const FR_NAME_NUMBER consistency_levels[] = {
	{ "any", CASS_CONSISTENCY_ANY },
	{ "one", CASS_CONSISTENCY_ONE },
	{ "two", CASS_CONSISTENCY_TWO },
	{ "three", CASS_CONSISTENCY_THREE },
	{ "quorum", CASS_CONSISTENCY_QUORUM },
	{ "all", CASS_CONSISTENCY_ALL },
	{ "each_quorum", CASS_CONSISTENCY_EACH_QUORUM },
	{ "local_quorum", CASS_CONSISTENCY_LOCAL_QUORUM },
	{ "local_one", CASS_CONSISTENCY_LOCAL_ONE },
	{ NULL, 0 }
	};

	void casslog(const CassLogMessage* message, void* data) {
	rlm_sql_cassandra_config_t *conf = data;
	int logPosition = __sync_fetch_and_add(&conf->logRingPosition, 1) % RLM_SQL_CASS_LOG_BUFFER;
	switch(message->severity) {
	case CASS_LOG_CRITICAL:
	case CASS_LOG_ERROR:
	conf->logBuffer[logPosition].type = L_ERR;
	break;
	case CASS_LOG_WARN:
	conf->logBuffer[logPosition].type = L_WARN;
	break;
	case CASS_LOG_INFO:
	case CASS_LOG_DISABLED:
	case CASS_LOG_LAST_ENTRY:
	conf->logBuffer[logPosition].type = L_INFO;
	break;
	case CASS_LOG_DEBUG:
	case CASS_LOG_TRACE:
	conf->logBuffer[logPosition].type = L_DBG;
	break;
	}
	if(conf->logBuffer[logPosition].msg != NULL) talloc_free(conf->logBuffer[logPosition].msg);
	MEM(conf->logBuffer[logPosition].msg = talloc_asprintf(conf, "rlm_sql_cassandra: [drv][%" PRId64 "][%d] In file %s, function %s, on line %d: %s", (int64_t)message->time_ms, (int)message->severity, message->file, message->function, message->line, message->message));
	}

	static int _mod_destructor(rlm_sql_cassandra_config_t *conf)
	{
	DEBUG2("rlm_sql_cassandra: Module shut down");
	cass_log_cleanup();
	return 0;
	}

	static int mod_instantiate(CONF_SECTION conf, rlm_sql_config_t config)
	{
	static bool version_done = false;

	rlm_sql_cassandra_config_t *driver;
	int consistency;

	if (!version_done) {
	version_done = true;

	INFO("rlm_sql_cassandra: compiled against Cassandra cpp-driver %d.%d.%d%s", CASS_VERSION_MAJOR, CASS_VERSION_MINOR, CASS_VERSION_PATCH, CASS_VERSION_SUFFIX);
	}

	MEM(driver = config->driver = talloc_zero(config, rlm_sql_cassandra_config_t));
	talloc_set_destructor(driver, _mod_destructor);

	if (cf_section_parse(conf, driver, driver_config) < 0) {
	return -1;
	}

	consistency = fr_str2int(consistency_levels, driver->consistency_str, -1);
	if (consistency < 0) {
	ERROR("rlm_sql_cassandra: Invalid consistency level %s", driver->consistency_str);
	return -1;
	}
	driver->consistency = (CassConsistency)consistency;
	driver->logRingPosition = 0;
	for(int i = 0; i < RLM_SQL_CASS_LOG_BUFFER; i++) {
	driver->logBuffer[i].msg = NULL;
	}
	cass_log_set_level(CASS_LOG_INFO);
	cass_log_set_callback(casslog, driver);

	return 0;
	}

	static int _sql_socket_destructor(rlm_sql_cassandra_conn_t *conn)
	{
	DEBUG2("rlm_sql_cassandra: Socket destructor called, closing socket");

	if (conn->resIter) {
	cass_iterator_free(conn->resIter);
	}
	if (conn->res) {
	cass_result_free(conn->res);
	}
	if (conn->sess) {
	cass_session_free(conn->sess);
	}
	if(conn->cluster) {
	cass_cluster_free(conn->cluster);
	}

	return 0;
	}

	static sql_rcode_t sql_socket_init(rlm_sql_handle_t handle, rlm_sql_config_t config)
	{
	rlm_sql_cassandra_conn_t *conn;
	CassCluster *cluster;
	CassSession *sess;
	CassFuture *future;
	CassError error;

	MEM(conn = handle->conn = talloc_zero(handle, rlm_sql_cassandra_conn_t));
	talloc_set_destructor(conn, _sql_socket_destructor);

	DEBUG4("rlm_sql_cassandra: Configuring CassCluster");
	cluster = conn->cluster = cass_cluster_new();
	cass_cluster_set_contact_points(cluster, config->sql_server);
	cass_cluster_set_port(cluster, atoi(config->sql_port));
	cass_cluster_set_connect_timeout(cluster, config->connect_timeout_ms);
	cass_cluster_set_request_timeout(cluster, config->query_timeout);
	cass_cluster_set_credentials(cluster, config->sql_login, config->sql_password);

	DEBUG4("rlm_sql_cassandra: Connecting to Cassandra Cluster");
	sess = conn->sess = cass_session_new();
	future = cass_session_connect_keyspace(sess, cluster, config->sql_db);
	error = cass_future_error_code(future);
	if(error != CASS_OK) {
	const char *msg; size_t msgLen;
	cass_future_error_message(future, &msg, &msgLen);
	ERROR("rlm_sql_cassandra: Unable to connect: [%x] %s", (int)error, msg);
	cass_future_free(future);
	return RLM_SQL_ERROR;
	}
	cass_future_free(future);

	DEBUG4("rlm_sql_cassandra: Connected to Cassandra Cluster");

	return RLM_SQL_OK;
	}

	static sql_rcode_t sql_query(rlm_sql_handle_t handle, rlm_sql_config_t config, char const *query)
	{
	rlm_sql_cassandra_conn_t *conn = handle->conn;
	rlm_sql_cassandra_config_t *conf = config->driver;
	CassStatement *stmt;
	CassFuture *future;
	CassError error;

	if(query[0] == ';') return RLM_SQL_OK;

	stmt = cass_statement_new(query, 0);
	cass_statement_set_consistency(stmt, conf->consistency);

	future = cass_session_execute(conn->sess, stmt);
	cass_statement_free(stmt);
	error = cass_future_error_code(future);
	if(error != CASS_OK) {
	const char *msg; size_t msgLen;
	cass_future_error_message(future, &msg, &msgLen);
	WARN("rlm_sql_cassandra: query failed: [%x] %s", (int)error, msg);
	DEBUG("rlm_sql_cassandra: the query was: %s", query);
	cass_future_free(future);
	switch(error) {
	case CASS_ERROR_SERVER_SYNTAX_ERROR:
	case CASS_ERROR_SERVER_INVALID_QUERY:
	return RLM_SQL_QUERY_INVALID;
	default:
	return RLM_SQL_ERROR;
	}
	}
	conn->res = cass_future_get_result(future);
	cass_future_free(future);
	if(!conn->res) DEBUG4("rlm_sql_cassandra: query doesn't return a result.");

	return RLM_SQL_OK;
	}

	static sql_rcode_t sql_store_result(rlm_sql_handle_t * handle, UNUSED rlm_sql_config_t *config)
	{
	return RLM_SQL_OK;
	}

	static int sql_num_fields(rlm_sql_handle_t handle, UNUSED rlm_sql_config_t config)
	{
	rlm_sql_cassandra_conn_t *conn = handle->conn;
	if(conn->res) {
	return cass_result_column_count(conn->res);
	}
	WARN("rlm_sql_cassandra: sql_num_fields called without an outstanding query result.");
	return 0;
	}

	static int sql_num_rows(rlm_sql_handle_t handle, UNUSED rlm_sql_config_t config)
	{
	rlm_sql_cassandra_conn_t *conn = handle->conn;
	if(conn->res) {
	return cass_result_row_count(conn->res);
	}
	WARN("rlm_sql_cassandra: sql_num_rows called without an outstanding query result.");
	return 0;
	}

	#define RLM_CASS_ERR_DATA_RETRIVE(type) {\
	const char *colName; size_t colNameLen;\
	if(cass_result_column_name(conn->res, i, &colName, &colNameLen) != CASS_OK) colName = "error";\
	WARN("rlm_sql_cassandra: failed to retrive " type " data at column (%d)%s", i, colName);\
	conn->resBuffer[i] = "";\
	break;}
	static sql_rcode_t sql_fetch_row(rlm_sql_handle_t handle, rlm_sql_config_t config)
	{
	rlm_sql_cassandra_conn_t *conn = handle->conn;
	handle->row = NULL;
	if(conn->res && !conn->resIter) {
	conn->resIter = cass_iterator_from_result(conn->res);
	}
	if(conn->res && conn->resIter) {
	if(cass_iterator_next(conn->resIter)) {
	if(conn->resBuffer != NULL) talloc_free(conn->resBuffer);
	const CassRow* row = cass_iterator_get_row(conn->resIter);
	int fields = sql_num_fields(handle, config);
	MEM(handle->row = conn->resBuffer = talloc_array(handle, char *, fields));
	for(int i = 0; i < fields; i++) {
	const CassValue *value = cass_row_get_column(row, i);
	conn->resBuffer[i] = NULL;
	if(cass_value_is_null(value) == cass_true) {
	continue;
	}
	CassValueType type = cass_value_type(value);
	switch(type) {
	case CASS_VALUE_TYPE_ASCII:
	case CASS_VALUE_TYPE_TEXT:
	case CASS_VALUE_TYPE_VARCHAR: {
	const char *str; size_t strLen;
	if(cass_value_get_string(value, &str, &strLen) != CASS_OK) {
	RLM_CASS_ERR_DATA_RETRIVE("string");
	}
	MEM(conn->resBuffer[i] = talloc_array(conn->resBuffer, char, strLen+1));
	memcpy(conn->resBuffer[i], str, strLen);
	conn->resBuffer[i][strLen] = '\0';
	break;
	}
	case CASS_VALUE_TYPE_BOOLEAN: {
	cass_bool_t bv;
	if(cass_value_get_bool(value, &bv) != CASS_OK) {
	RLM_CASS_ERR_DATA_RETRIVE("bool");
	}
	MEM(conn->resBuffer[i] = talloc_zero_array(conn->resBuffer, char, 2));
	if(bv == cass_false) {
	conn->resBuffer[i][0] = '0';
	} else {
	conn->resBuffer[i][0] = '1';
	}
	break;
	}
	case CASS_VALUE_TYPE_INT: {
	cass_int32_t i32v;
	if(cass_value_get_int32(value, &i32v) != CASS_OK) {
	RLM_CASS_ERR_DATA_RETRIVE("int32");
	}
	MEM(conn->resBuffer[i] = talloc_asprintf(conn->resBuffer, "%"PRId32, (int32_t)i32v));
	break;
	}
	case CASS_VALUE_TYPE_TIMESTAMP:
	case CASS_VALUE_TYPE_BIGINT: {
	cass_int64_t i64v;
	if(cass_value_get_int64(value, &i64v) != CASS_OK) {
	RLM_CASS_ERR_DATA_RETRIVE("int64");
	}
	MEM(conn->resBuffer[i] = talloc_asprintf(conn->resBuffer, "%"PRId64, (int64_t)i64v));
	break;
	}
	case CASS_VALUE_TYPE_UUID:
	case CASS_VALUE_TYPE_TIMEUUID: {
	CassUuid uuid;
	if(cass_value_get_uuid(value, &uuid) != CASS_OK) {
	RLM_CASS_ERR_DATA_RETRIVE("UUID");
	}
	MEM(conn->resBuffer[i] = talloc_array(conn->resBuffer, char, CASS_UUID_STRING_LENGTH));
	cass_uuid_string(uuid, conn->resBuffer[i]);
	break;
	}
	default: {
	const char *colName; size_t colNameLen;
	if(cass_result_column_name(conn->res, i, &colName, &colNameLen) != CASS_OK) colName = "error";
	WARN("rlm_sql_cassandra: column (%d)%s: Unsupported type %d", i, colName, (int)type);
	break;
	}
	}
	DEBUG4("rlm_sql_cassandra: row[%d] => %s", i, conn->resBuffer[i] ? conn->resBuffer[i] : "(null)");
	}
	} else DEBUG4("rlm_sql_cassandra: no more results.");
	} else WARN("rlm_sql_cassandra: sql_fetch_row called without an outstanding query result.");
	return RLM_SQL_OK;
	}

	static sql_rcode_t sql_free_result(rlm_sql_handle_t * handle, UNUSED rlm_sql_config_t *config)
	{
	rlm_sql_cassandra_conn_t *conn = handle->conn;

	if(conn->resBuffer != NULL) talloc_free(conn->resBuffer);
	if(conn->resIter != NULL) cass_iterator_free(conn->resIter);
	if(conn->res != NULL) cass_result_free(conn->res);

	conn->resBuffer = NULL;
	conn->resIter = NULL;
	conn->res = NULL;

	return RLM_SQL_OK;
	}

	static size_t sql_error(TALLOC_CTX ctx, sql_log_entry_t out[], size_t outlen, UNUSED rlm_sql_handle_t handle, rlm_sql_config_t *config)
	{
	rlm_sql_cassandra_config_t *driver = config->driver;
	int fillSize = outlen > RLM_SQL_CASS_LOG_BUFFER ? RLM_SQL_CASS_LOG_BUFFER : outlen;
	int bufferFilled = driver->logRingPosition;
	fillSize = fillSize > bufferFilled ? bufferFilled : fillSize;

	for(int i = 0; i < fillSize; i++) {
	sql_log_entry_t entry = driver->logBuffer[(bufferFilled - fillSize + i) % RLM_SQL_CASS_LOG_BUFFER];
	size_t msgLen = strlen(entry.msg)+1;
	out[i].type = entry.type;
	out[i].msg = talloc_array(ctx, char, msgLen);
	memcpy(out[i].msg, entry.msg, msgLen);
	}

	return fillSize;
	}

	static sql_rcode_t sql_finish_query(rlm_sql_handle_t handle, rlm_sql_config_t config)
	{
	return sql_free_result(handle, config);
	}

	//Unsupported operation on Cassandra Cluster, fake one.
	static int sql_affected_rows(UNUSED rlm_sql_handle_t * handle, UNUSED rlm_sql_config_t *config) { return 1; }

	/* Exported to rlm_sql */
	extern rlm_sql_module_t rlm_sql_cassandra;
	rlm_sql_module_t rlm_sql_cassandra = {
	.name = "rlm_sql_cassandra",
	.mod_instantiate = mod_instantiate,
	.sql_socket_init = sql_socket_init,
	.sql_query = sql_query,
	.sql_select_query = sql_query,
	.sql_store_result = sql_store_result,
	.sql_num_fields = sql_num_fields,
	.sql_num_rows = sql_num_rows,
	.sql_fetch_row = sql_fetch_row,
	.sql_free_result = sql_free_result,
	.sql_error = sql_error,
	.sql_finish_query = sql_finish_query,
	.sql_finish_select_query = sql_finish_query,
	.sql_affected_rows = sql_affected_rows
	};
	CREATE KEYSPACE radius WITH REPLICATION = { 'class' : 'SimpleStrategy', 'replication_factor' : 1 };
	USE radius;

	CREATE TABLE nas (
	id uuid PRIMARY KEY,
	nasname text,
	shortname text,
	type text,
	secret text,
	server text
	);

	CREATE TABLE radpostauth (
	username text,
	authdate bigint,
	reply text,
	PRIMARY KEY (username, authdate)
	) WITH CLUSTERING ORDER BY (authdate ASC);

	CREATE TABLE radusergroup (
	username text,
	priority int,
	groupname text,
	PRIMARY KEY (username, priority)
	) WITH CLUSTERING ORDER BY (priority ASC);

	CREATE TABLE radreply (
	id uuid,
	username text,
	attribute text,
	op text,
	value text,
	PRIMARY KEY (username, attribute)
	);

	CREATE TABLE radgroupreply (
	id uuid,
	groupname text,
	attribute text,
	op text,
	value text,
	PRIMARY KEY (groupname, attribute)
	);

	CREATE TABLE radgroupcheck (
	id uuid,
	groupname text,
	attribute text,
	op text,
	value text,
	PRIMARY KEY (groupname, attribute)
	);

	CREATE TABLE radcheck (
	id uuid,
	username text,
	attribute text,
	op text,
	value text,
	PRIMARY KEY (username, attribute)
	);

	CREATE TABLE radacct (
	id uuid,
	sessionid text,
	username text,
	realm text,
	nasipaddress text,
	nasportid text,
	nasporttype text,
	starttime bigint,
	updatetime bigint,
	stoptime bigint,
	sessiontime int,
	authentic text,
	connectinfo_start text,
	connectinfo_stop text,
	inputoctetslo bigint,
	inputoctetshi bigint,
	outputoctetslo bigint,
	outputoctetshi bigint,
	calledstationid text,
	callingstationid text,
	servicetype text,
	terminatecause text,
	framedprotocol text,
	framedipaddress text,
	PRIMARY KEY (username, nasipaddress, sessionid)
	);

	CREATE TABLE radnasflip (
	nasipaddress text,
	timestamp bigint,
	PRIMARY KEY (timestamp, nasipaddress)
	);

	CREATE TABLE radsess (
	id uuid,
	sessionid text,
	username text,
	nasipaddress text,
	nasportid text,
	callingstationid text,
	framedprotocol text,
	framedipaddress text,
	PRIMARY KEY (nasipaddress, username, sessionid)
	);
	CREATE INDEX ON radsess(username);
	sql cassauth {
	driver = "rlm_sql_cassandra"
	server = "localhost"
	port = 9042
	login = ""
	password = ""
	radius_db = "radius"
	query_timeout = 15000
	safe_characters = "@abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789.-_: /[]"
	sql_user_name = "%{%{Stripped-User-Name}:-%{%{User-Name}:-Guest}}"
	default_user_profile = "DEFAULT"

	cassandra {
	consistency = "local_quorum"
	}

	pool {
	start = 1
	min = 1
	max = ${thread[pool].max_servers}
	spare = 2
	uses = 0
	lifetime = 0
	idle_timeout = 60
	}

	read_clients = no
	postauth_table = "radpostauth"
	authcheck_table = "radcheck"
	groupcheck_table = "radgroupcheck"
	authreply_table = "radreply"
	groupreply_table = "radgroupreply"
	usergroup_table = "radusergroup"
	client_table = "nas"
	session_table = "radsess"
	session_ttl = "USING TTL 3600"
	postauth_ttl = "USING TTL 6307200"

	simul_count_query = "\
	SELECT COUNT(*) \
	FROM ${session_table} \
	WHERE username = '%{SQL-User-Name}'"

	simul_verify_query = "\
	SELECT id, sessionid, username, nasipaddress, nasportid, framedipaddress, \
	callingstationid, framedprotocol \
	FROM ${session_table} \
	WHERE username = '%{SQL-User-Name}'"

	client_query = "\
	SELECT id, nasname, shortname, type, secret, server \
	FROM ${client_table}"

	authorize_check_query = "\
	SELECT id, username, attribute, value, op \
	FROM ${authcheck_table} \
	WHERE username = '%{SQL-User-Name}'"

	authorize_reply_query = "\
	SELECT id, username, attribute, value, op \
	FROM ${authreply_table} \
	WHERE username = '%{SQL-User-Name}'"

	group_membership_query = "\
	SELECT groupname \
	FROM ${usergroup_table} \
	WHERE username = '%{SQL-User-Name}' \
	ORDER BY priority"

	authorize_group_check_query = "\
	SELECT id, groupname, attribute, value, op \
	FROM ${groupcheck_table} \
	WHERE groupname = '%{Sql-Group}'"

	authorize_group_reply_query = "\
	SELECT id, groupname, attribute, value, op \
	FROM ${groupreply_table} \
	WHERE groupname = '%{Sql-Group}'"

	accounting {
	reference = "%{tolower:type.%{Acct-Status-Type}.query}"

	type {
	accounting-on {
	query = "DELETE FROM ${....session_table} WHERE nasipaddress = '%{NAS-IP-Address}'"
	}

	accounting-off {
	query = "${..accounting-on.query}"
	}

	start {
	query = "\
	INSERT INTO ${....session_table} ( \
	id, sessionid, username, \
	nasipaddress, nasportid, \
	callingstationid, \
	framedprotocol, framedipaddress \
	) VALUES ( \
	uuid(), '%{Acct-Session-Id}', '%{SQL-User-Name}', \
	'%{NAS-IP-Address}', '%{NAS-Port}', \
	'%{Calling-Station-Id}', \
	'%{Framed-Protocol}', '%{Framed-IP-Address}' \
	) ${....session_ttl};"
	}

	interim-update {
	query = "${..start.query}"
	}

	stop {
	query = "DELETE FROM ${....session_table} \
	WHERE nasipaddress = '%{NAS-IP-Address}' \
	AND username = '%{SQL-User-Name}' \
	AND sessionid = '%{Acct-Session-Id}'"
	}
	}
	}

	post-auth {
	query = "\
	INSERT INTO ${..postauth_table} \
	(username, reply, authdate) \
	VALUES ( \
	'%{SQL-User-Name}', \
	'%{reply:Packet-Type}', \
	unixTimestampOf(now()) \
	) ${..postauth_ttl};"
	}
	}

	sql cassacct {
	driver = "rlm_sql_cassandra"
	server = "localhost"
	port = 9042
	login = ""
	password = ""
	radius_db = "radius"
	query_timeout = 15000
	safe_characters = "@abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789.-_: /[]"
	sql_user_name = "%{%{Stripped-User-Name}:-%{%{User-Name}:-Guest}}"
	default_user_profile = "DEFAULT"

	cassandra {
	consistency = "local_quorum"
	}

	pool {
	start = 1
	min = 1
	max = ${thread[pool].max_servers}
	spare = 2
	uses = 0
	lifetime = 0
	idle_timeout = 60
	}

	acct_ttl = "USING TTL 6307200"
	acct_table = "radacct"
	flip_table = "radnasflip"

	accounting {
	reference = "%{tolower:type.%{Acct-Status-Type}.query}"

	type {
	accounting-on {
	query = "\
	INSERT INTO ${....flip_table} (nasipaddress, timestamp) \
	VALUES ('%{NAS-IP-Address}', %{integer:Event-Timestamp}) \
	${....acct_ttl};"
	}

	accounting-off {
	query = "${..accounting-on.query}"
	}

	start {
	query = "\
	INSERT INTO ${....acct_table} ( \
	id, sessionid, username, realm, \
	nasipaddress, nasportid, nasporttype, \
	starttime, updatetime, stoptime, sessiontime, \
	authentic, connectinfo_start, connectinfo_stop, \
	inputoctetslo, inputoctetshi, outputoctetslo, outputoctetshi, \
	calledstationid, callingstationid, servicetype, \
	terminatecause, framedprotocol, framedipaddress \
	) VALUES ( \
	uuid(), '%{Acct-Session-Id}', '%{SQL-User-Name}', '%{Realm}', \
	'%{NAS-IP-Address}', '%{NAS-Port}', '%{NAS-Port-Type}', \
	%{integer:Event-Timestamp}, %{integer:Event-Timestamp}, \
	0, 0, \
	'%{Acct-Authentic}', '%{Connect-Info}', '', \
	0, 0, 0, 0, \
	'%{Called-Station-Id}', '%{Calling-Station-Id}', '%{Service-Type}', \
	'', '%{Framed-Protocol}', '%{Framed-IP-Address}' \
	) ${....acct_ttl};"
	}

	interim-update {
	query = "\
	INSERT INTO ${....acct_table} ( \
	sessionid, username, nasipaddress, \
	updatetime, sessiontime, \
	framedipaddress, \
	inputoctetshi, inputoctetslo, \
	outputoctetshi, outputoctetslo \
	) VALUES ( \
	'%{Acct-Session-Id}', '%{SQL-User-Name}', '%{NAS-IP-Address}', \
	%{integer:Event-Timestamp}, %{Acct-Session-Time}, \
	'%{Framed-IP-Address}', \
	%{%{Acct-Input-Gigawords}:-0}, %{%{Acct-Input-Octets}:-0}, \
	%{%{Acct-Output-Gigawords}:-0}, %{%{Acct-Output-Octets}:-0} \
	) ${....acct_ttl};"
	}

	stop {
	query = "\
	INSERT INTO ${....acct_table} ( \
	sessionid, username, nasipaddress, \
	updatetime, sessiontime, stoptime, \
	inputoctetshi, inputoctetslo, \
	outputoctetshi, outputoctetslo, \
	connectinfo_stop, terminatecause \
	) VALUES ( \
	'%{Acct-Session-Id}', '%{SQL-User-Name}', '%{NAS-IP-Address}', \
	%{integer:Event-Timestamp}, %{Acct-Session-Time}, %{integer:Event-Timestamp}, \
	%{%{Acct-Input-Gigawords}:-0}, %{%{Acct-Input-Octets}:-0}, \
	%{%{Acct-Output-Gigawords}:-0}, %{%{Acct-Output-Octets}:-0}, \
	'%{Connect-Info}', '%{Acct-Terminate-Cause}' \
	) ${....acct_ttl};"
	}
	}
	}
	}

	sql cassacctsupplement {
	driver = "rlm_sql_cassandra"
	server = "localhost"
	port = 9042
	login = ""
	password = ""
	radius_db = "radius"
	query_timeout = 15000
	safe_characters = "@abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789.-_: /[]"
	sql_user_name = "%{%{Stripped-User-Name}:-%{%{User-Name}:-Guest}}"
	default_user_profile = "DEFAULT"

	cassandra {
	consistency = "local_quorum"
	}

	pool {
	start = 1
	min = 1
	max = ${thread[pool].max_servers}
	spare = 2
	uses = 0
	lifetime = 0
	idle_timeout = 60
	}

	acct_ttl = "USING TTL 6307200"
	acct_table = "radacct"

	accounting {
	reference = "%{tolower:type.%{Acct-Status-Type}.query}"

	type {
	interim-update {
	query = "\
	INSERT INTO radacct ( \
	id, sessionid, username, realm, \
	nasipaddress, nasportid, nasporttype, \
	starttime, stoptime, \
	authentic, connectinfo_start, connectinfo_stop, \
	calledstationid, callingstationid, servicetype, \
	terminatecause, framedprotocol \
	) VALUES ( \
	uuid(), '%{Acct-Session-Id}', '%{SQL-User-Name}', '%{Realm}', \
	'%{NAS-IP-Address}', '%{NAS-Port}', '%{NAS-Port-Type}', \
	0, 0, \
	'%{Acct-Authentic}', '%{Connect-Info}', '', \
	'%{Called-Station-Id}', '%{Calling-Station-Id}', '%{Service-Type}', \
	'', '%{Framed-Protocol}' \
	) IF NOT EXISTS ${....acct_ttl};"
	}

	stop {
	query = "\
	INSERT INTO radacct ( \
	id, sessionid, username, realm, \
	nasipaddress, nasportid, nasporttype, \
	starttime, \
	authentic, connectinfo_start, \
	calledstationid, callingstationid, servicetype, \
	framedprotocol, framedipaddress \
	) VALUES ( \
	uuid(), '%{Acct-Session-Id}', '%{SQL-User-Name}', '%{Realm}', \
	'%{NAS-IP-Address}', '%{NAS-Port}', '%{NAS-Port-Type}', \
	0, \
	'%{Acct-Authentic}', '', \
	'%{Called-Station-Id}', '%{Calling-Station-Id}', '%{Service-Type}', \
	'%{Framed-Protocol}', '%{Framed-IP-Address}' \
	) IF NOT EXISTS ${....acct_ttl};"
	}

	start {
	query = ";"
	}
	accounting-on {
	query = ";"
	}
	accounting-off {
	query = ";"
	}
	}
	}
	}