Skip to content

Commit

Permalink
RANGER-246 - Updates based on changes on Kafka side
Browse files Browse the repository at this point in the history
  • Loading branch information
@boscodurai
boscodurai committed May 24, 2015
1 parent f31274b commit 115577a
Show file tree
Hide file tree
Showing 8 changed files with 404 additions and 282 deletions.
4 changes: 4 additions & 0 deletions agents-common/scripts/enable-agent.sh
View file
Original file line number Diff line number Diff line change
Expand Up @@ -151,6 +151,8 @@ if [ "${HCOMPONENT_NAME}" = "knox" ]; then
HCOMPONENT_LIB_DIR=${HCOMPONENT_INSTALL_DIR}/ext
elif [ "${HCOMPONENT_NAME}" = "solr" ]; then
HCOMPONENT_LIB_DIR=${HCOMPONENT_INSTALL_DIR}/solr-webapp/webapp/WEB-INF/lib
elif [ "${HCOMPONENT_NAME}" = "kafka" ]; then
HCOMPONENT_LIB_DIR=${HCOMPONENT_INSTALL_DIR}/libs
fi

HCOMPONENT_CONF_DIR=${HCOMPONENT_INSTALL_DIR}/conf
Expand All @@ -163,6 +165,8 @@ if [ "${HCOMPONENT_NAME}" = "solr" ]; then
echo "INFO: Changing ownership of $HCOMPONENT_CONF_DIR to $install_owner"
chown $install_owner:$install_owner $HCOMPONENT_CONF_DIR
fi
elif [ "${HCOMPONENT_NAME}" = "kafka" ]; then
HCOMPONENT_CONF_DIR=${HCOMPONENT_INSTALL_DIR}/config
fi

HCOMPONENT_ARCHIVE_CONF_DIR=${HCOMPONENT_CONF_DIR}/.archive
Expand Down
200 changes: 132 additions & 68 deletions agents-common/src/main/java/org/apache/ranger/plugin/service/RangerBasePlugin.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -20,16 +20,15 @@
package org.apache.ranger.plugin.service;

import java.util.Collection;
import java.util.List;
import java.util.Hashtable;
import java.util.Map;

import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.ranger.admin.client.RangerAdminClient;
import org.apache.ranger.admin.client.RangerAdminRESTClient;
import org.apache.ranger.authorization.hadoop.config.RangerConfiguration;
import org.apache.ranger.plugin.contextenricher.RangerContextEnricher;
import org.apache.ranger.plugin.model.RangerServiceDef;
import org.apache.ranger.plugin.policyengine.RangerAccessRequest;
import org.apache.ranger.plugin.policyengine.RangerAccessRequestImpl;
Expand All @@ -44,22 +43,23 @@
import org.apache.ranger.plugin.util.PolicyRefresher;
import org.apache.ranger.plugin.util.ServicePolicies;


public class RangerBasePlugin {
private static final Log LOG = LogFactory.getLog(RangerBasePlugin.class);

private String serviceType = null;
private String appId = null;
private String serviceName = null;
private PolicyRefresher refresher = null;
private RangerPolicyEngine policyEngine = null;
private String serviceType = null;
private String appId = null;
private String serviceName = null;
private PolicyRefresher refresher = null;
private RangerPolicyEngine policyEngine = null;
private RangerPolicyEngineOptions policyEngineOptions = new RangerPolicyEngineOptions();
private RangerAccessResultProcessor resultProcessor = null;

Map<String, LogHistory> logHistoryList = new Hashtable<String, RangerBasePlugin.LogHistory>();
int logInterval = 30000; // 30 seconds

public RangerBasePlugin(String serviceType, String appId) {
this.serviceType = serviceType;
this.appId = appId;
this.appId = appId;
}

public String getServiceType() {
Expand All @@ -75,7 +75,8 @@ public RangerServiceDef getServiceDef() {
public int getServiceDefId() {
RangerServiceDef serviceDef = getServiceDef();

return serviceDef != null && serviceDef.getId() != null ? serviceDef.getId().intValue() : -1;
return serviceDef != null && serviceDef.getId() != null ? serviceDef
.getId().intValue() : -1;
}

public String getAppId() {
Expand All @@ -89,41 +90,62 @@ public String getServiceName() {
public void init() {
cleanup();

RangerConfiguration.getInstance().addResourcesForServiceType(serviceType);
RangerConfiguration.getInstance().addResourcesForServiceType(
serviceType);
RangerConfiguration.getInstance().initAudit(appId);

String propertyPrefix = "ranger.plugin." + serviceType;
long pollingIntervalMs = RangerConfiguration.getInstance().getLong(propertyPrefix + ".policy.pollIntervalMs", 30 * 1000);
String cacheDir = RangerConfiguration.getInstance().get(propertyPrefix + ".policy.cache.dir");

serviceName = RangerConfiguration.getInstance().get(propertyPrefix + ".service.name");

policyEngineOptions.evaluatorType = RangerConfiguration.getInstance().get(propertyPrefix + ".policyengine.option.evaluator.type", RangerPolicyEvaluator.EVALUATOR_TYPE_CACHED);
policyEngineOptions.cacheAuditResults = RangerConfiguration.getInstance().getBoolean(propertyPrefix + ".policyengine.option.cache.audit.results", true);
policyEngineOptions.disableContextEnrichers = RangerConfiguration.getInstance().getBoolean(propertyPrefix + ".policyengine.option.disable.context.enrichers", false);
policyEngineOptions.disableCustomConditions = RangerConfiguration.getInstance().getBoolean(propertyPrefix + ".policyengine.option.disable.custom.conditions", false);

String propertyPrefix = "ranger.plugin." + serviceType;
long pollingIntervalMs = RangerConfiguration.getInstance().getLong(
propertyPrefix + ".policy.pollIntervalMs", 30 * 1000);
String cacheDir = RangerConfiguration.getInstance().get(
propertyPrefix + ".policy.cache.dir");

serviceName = RangerConfiguration.getInstance().get(
propertyPrefix + ".service.name");

policyEngineOptions.evaluatorType = RangerConfiguration.getInstance()
.get(propertyPrefix + ".policyengine.option.evaluator.type",
RangerPolicyEvaluator.EVALUATOR_TYPE_CACHED);
policyEngineOptions.cacheAuditResults = RangerConfiguration
.getInstance().getBoolean(
propertyPrefix
+ ".policyengine.option.cache.audit.results",
true);
policyEngineOptions.disableContextEnrichers = RangerConfiguration
.getInstance()
.getBoolean(
propertyPrefix
+ ".policyengine.option.disable.context.enrichers",
false);
policyEngineOptions.disableCustomConditions = RangerConfiguration
.getInstance()
.getBoolean(
propertyPrefix
+ ".policyengine.option.disable.custom.conditions",
false);

RangerAdminClient admin = createAdminClient(propertyPrefix);

refresher = new PolicyRefresher(this, serviceType, appId, serviceName, admin, pollingIntervalMs, cacheDir);
refresher = new PolicyRefresher(this, serviceType, appId, serviceName,
admin, pollingIntervalMs, cacheDir);
refresher.startRefresher();
}

public void setPolicies(ServicePolicies policies) {
RangerPolicyEngine policyEngine = new RangerPolicyEngineImpl(policies, policyEngineOptions);
RangerPolicyEngine policyEngine = new RangerPolicyEngineImpl(policies,
policyEngineOptions);

this.policyEngine = policyEngine;
}

public void cleanup() {
PolicyRefresher refresher = this.refresher;

this.serviceName = null;
this.serviceName = null;
this.policyEngine = null;
this.refresher = null;
this.refresher = null;

if(refresher != null) {
if (refresher != null) {
refresher.stopRefresher();
}
}
Expand All @@ -140,14 +162,16 @@ public RangerAccessResult isAccessAllowed(RangerAccessRequest request) {
return isAccessAllowed(request, resultProcessor);
}

public Collection<RangerAccessResult> isAccessAllowed(Collection<RangerAccessRequest> requests) {
public Collection<RangerAccessResult> isAccessAllowed(
Collection<RangerAccessRequest> requests) {
return isAccessAllowed(requests, resultProcessor);
}

public RangerAccessResult isAccessAllowed(RangerAccessRequest request, RangerAccessResultProcessor resultProcessor) {
public RangerAccessResult isAccessAllowed(RangerAccessRequest request,
RangerAccessResultProcessor resultProcessor) {
RangerPolicyEngine policyEngine = this.policyEngine;

if(policyEngine != null) {
if (policyEngine != null) {
policyEngine.enrichContext(request);

return policyEngine.isAccessAllowed(request, resultProcessor);
Expand All @@ -156,10 +180,12 @@ public RangerAccessResult isAccessAllowed(RangerAccessRequest request, RangerAcc
return null;
}

public Collection<RangerAccessResult> isAccessAllowed(Collection<RangerAccessRequest> requests, RangerAccessResultProcessor resultProcessor) {
public Collection<RangerAccessResult> isAccessAllowed(
Collection<RangerAccessRequest> requests,
RangerAccessResultProcessor resultProcessor) {
RangerPolicyEngine policyEngine = this.policyEngine;

if(policyEngine != null) {
if (policyEngine != null) {
policyEngine.enrichContext(requests);

return policyEngine.isAccessAllowed(requests, resultProcessor);
Expand All @@ -171,24 +197,26 @@ public Collection<RangerAccessResult> isAccessAllowed(Collection<RangerAccessReq
public RangerAccessResult createAccessResult(RangerAccessRequest request) {
RangerPolicyEngine policyEngine = this.policyEngine;

if(policyEngine != null) {
if (policyEngine != null) {
return policyEngine.createAccessResult(request);
}

return null;
}

public void grantAccess(GrantRevokeRequest request, RangerAccessResultProcessor resultProcessor) throws Exception {
if(LOG.isDebugEnabled()) {
public void grantAccess(GrantRevokeRequest request,
RangerAccessResultProcessor resultProcessor) throws Exception {
if (LOG.isDebugEnabled()) {
LOG.debug("==> RangerAdminRESTClient.grantAccess(" + request + ")");
}

PolicyRefresher refresher = this.refresher;
RangerAdminClient admin = refresher == null ? null : refresher.getRangerAdminClient();
boolean isSuccess = false;
PolicyRefresher refresher = this.refresher;
RangerAdminClient admin = refresher == null ? null : refresher
.getRangerAdminClient();
boolean isSuccess = false;

try {
if(admin == null) {
if (admin == null) {
throw new Exception("ranger-admin client is null");
}

Expand All @@ -199,22 +227,24 @@ public void grantAccess(GrantRevokeRequest request, RangerAccessResultProcessor
auditGrantRevoke(request, "grant", isSuccess, resultProcessor);
}

if(LOG.isDebugEnabled()) {
if (LOG.isDebugEnabled()) {
LOG.debug("<== RangerAdminRESTClient.grantAccess(" + request + ")");
}
}

public void revokeAccess(GrantRevokeRequest request, RangerAccessResultProcessor resultProcessor) throws Exception {
if(LOG.isDebugEnabled()) {
public void revokeAccess(GrantRevokeRequest request,
RangerAccessResultProcessor resultProcessor) throws Exception {
if (LOG.isDebugEnabled()) {
LOG.debug("==> RangerAdminRESTClient.revokeAccess(" + request + ")");
}

PolicyRefresher refresher = this.refresher;
RangerAdminClient admin = refresher == null ? null : refresher.getRangerAdminClient();
boolean isSuccess = false;
PolicyRefresher refresher = this.refresher;
RangerAdminClient admin = refresher == null ? null : refresher
.getRangerAdminClient();
boolean isSuccess = false;

try {
if(admin == null) {
if (admin == null) {
throw new Exception("ranger-admin client is null");
}

Expand All @@ -225,59 +255,72 @@ public void revokeAccess(GrantRevokeRequest request, RangerAccessResultProcessor
auditGrantRevoke(request, "revoke", isSuccess, resultProcessor);
}

if(LOG.isDebugEnabled()) {
if (LOG.isDebugEnabled()) {
LOG.debug("<== RangerAdminRESTClient.revokeAccess(" + request + ")");
}
}


private RangerAdminClient createAdminClient(String propertyPrefix) {
if(LOG.isDebugEnabled()) {
LOG.debug("==> RangerAdminRESTClient.createAdminClient(" + propertyPrefix + ")");
if (LOG.isDebugEnabled()) {
LOG.debug("==> RangerAdminRESTClient.createAdminClient("
+ propertyPrefix + ")");
}

RangerAdminClient ret = null;

String propertyName = propertyPrefix + ".policy.source.impl";
String policySourceImpl = RangerConfiguration.getInstance().get(propertyName);
String policySourceImpl = RangerConfiguration.getInstance().get(
propertyName);

if(StringUtils.isEmpty(policySourceImpl)) {
if (StringUtils.isEmpty(policySourceImpl)) {
if (LOG.isDebugEnabled()) {
LOG.debug(String.format("Value for property[%s] was null or empty. Unxpected! Will use policy source of type[%s]", propertyName, RangerAdminRESTClient.class.getName()));
LOG.debug(String
.format("Value for property[%s] was null or empty. Unxpected! Will use policy source of type[%s]",
propertyName,
RangerAdminRESTClient.class.getName()));
}
} else {
if (LOG.isDebugEnabled()) {
LOG.debug(String.format("Value for property[%s] was [%s].", propertyName, policySourceImpl));
LOG.debug(String.format("Value for property[%s] was [%s].",
propertyName, policySourceImpl));
}
try {
@SuppressWarnings("unchecked")
Class<RangerAdminClient> adminClass = (Class<RangerAdminClient>)Class.forName(policySourceImpl);

Class<RangerAdminClient> adminClass = (Class<RangerAdminClient>) Class
.forName(policySourceImpl);

ret = adminClass.newInstance();
} catch (Exception excp) {
LOG.error("failed to instantiate policy source of type '" + policySourceImpl + "'. Will use policy source of type '" + RangerAdminRESTClient.class.getName() + "'", excp);
LOG.error("failed to instantiate policy source of type '"
+ policySourceImpl
+ "'. Will use policy source of type '"
+ RangerAdminRESTClient.class.getName() + "'", excp);
}
}

if(ret == null) {
if (ret == null) {
ret = new RangerAdminRESTClient();
}

ret.init(serviceName, appId, propertyPrefix);

if(LOG.isDebugEnabled()) {
LOG.debug("<== RangerAdminRESTClient.createAdminClient(" + propertyPrefix + "): policySourceImpl=" + policySourceImpl + ", client=" + ret);
if (LOG.isDebugEnabled()) {
LOG.debug("<== RangerAdminRESTClient.createAdminClient("
+ propertyPrefix + "): policySourceImpl="
+ policySourceImpl + ", client=" + ret);
}
return ret;
}

private void auditGrantRevoke(GrantRevokeRequest request, String action, boolean isSuccess, RangerAccessResultProcessor resultProcessor) {
private void auditGrantRevoke(GrantRevokeRequest request, String action,
boolean isSuccess, RangerAccessResultProcessor resultProcessor) {
RangerPolicyEngine policyEngine = this.policyEngine;

if(request != null && resultProcessor != null && policyEngine != null) {
if (request != null && resultProcessor != null && policyEngine != null) {
RangerAccessRequestImpl accessRequest = new RangerAccessRequestImpl();

accessRequest.setResource(new RangerAccessResourceImpl(request.getResource()));

accessRequest.setResource(new RangerAccessResourceImpl(request
.getResource()));
accessRequest.setUser(request.getGrantor());
accessRequest.setAccessType(RangerPolicyEngine.ADMIN_ACCESS);
accessRequest.setAction(action);
Expand All @@ -287,18 +330,39 @@ private void auditGrantRevoke(GrantRevokeRequest request, String action, boolean
accessRequest.setSessionId(request.getSessionId());

// call isAccessAllowed() to determine if audit is enabled or not
RangerAccessResult accessResult = policyEngine.isAccessAllowed(accessRequest, null);
RangerAccessResult accessResult = policyEngine.isAccessAllowed(
accessRequest, null);

if(accessResult != null && accessResult.getIsAudited()) {
if (accessResult != null && accessResult.getIsAudited()) {
accessRequest.setAccessType(action);
accessResult.setIsAllowed(isSuccess);

if(! isSuccess) {
if (!isSuccess) {
accessResult.setPolicyId(-1);
}

resultProcessor.processResult(accessResult);
}
}
}

public boolean logErrorMessage(String message) {
LogHistory log = logHistoryList.get(message);
if (log == null) {
log = new LogHistory();
log.message = message;
log.lastLogTime = 0;
}
if ((System.currentTimeMillis() - log.lastLogTime) > logInterval) {
log.lastLogTime = System.currentTimeMillis();
LOG.error(message);
return true;
}
return false;
}

class LogHistory {
long lastLogTime;
String message;
}
}
Loading

0 comments on commit 115577a

Please sign in to comment.